"""Canonical contract normalizer for PMFlow AI turn runtime. This module is the boundary between imperfect goal-assessor output and the runtime loop. The loop should only consume canonical GoalFrame / OperationGap contracts, not legacy route labels or free-form tool instructions. """ from __future__ import annotations from copy import deepcopy from typing import Any from ..schemas.pmflow_ai_turn_runtime import ( AnswerShape, DisclosureResponseMode, DisclosureVisibilityClass, OperationKind, PMFlowAIGoalFrame, PMFlowAIOperationGap, PMFlowAITurnState, RuntimeActionKind, SourcePolicy, ) CANONICAL_RUNTIME_ACTIONS: set[str] = { "inspect_surface_context", "inspect_internal_projects", "inspect_saved_ideas", "resolve_project_objects", "inspect_project_status", "external_search", "external_fetch", "synthesize_answer", "ask_clarification", } READ_RUNTIME_ACTIONS: set[str] = CANONICAL_RUNTIME_ACTIONS - {"synthesize_answer", "ask_clarification"} INTERNAL_READ_ACTIONS: set[str] = { "inspect_surface_context", "inspect_internal_projects", "inspect_saved_ideas", "resolve_project_objects", "inspect_project_status", } EXTERNAL_READ_ACTIONS: set[str] = {"external_search", "external_fetch"} USER_GAP_TYPES = {"user_clarification_gap", "user_authorization_gap"} SYSTEM_GAP_TYPES = {"system_evidence_gap", "freshness_gap"} DISCLOSURE_VISIBILITY_CLASSES: set[str] = { "public_product_capability", "tenant_business_state", "external_public", "protected_pmflow_internal", "platform_admin_debug", } DISCLOSURE_RESPONSE_MODES: set[str] = { "answer", "safe_product_boundary", "protected_refusal", "platform_admin_redirect", } SOURCE_SCOPE_BY_OPERATION: dict[str, tuple[str, ...]] = { "explain_assistant": ("internal", "none"), "list_collection": ("internal",), "inspect_object": ("internal",), "guide_execution": ("internal", "hybrid"), "judge_direction": ("external", "hybrid", "none"), "compare_objects": ("external", "hybrid"), "answer_general": ("internal", "external", "hybrid", "none"), } ANSWER_SHAPES_BY_OPERATION: dict[str, tuple[str, ...]] = { "explain_assistant": ("general_answer",), "list_collection": ("fact_list",), "inspect_object": ("status_answer", "execution_guidance", "composite_answer"), "guide_execution": ("execution_guidance", "status_answer", "composite_answer"), "judge_direction": ("strategic_judgement", "composite_answer", "general_answer"), "compare_objects": ("strategic_judgement", "composite_answer"), "answer_general": ( "general_answer", "strategic_judgement", "status_answer", "clarification", "execution_guidance", "composite_answer", ), } def normalize_turn_hypothesis_contract(hypothesis: dict[str, Any] | None) -> dict[str, Any]: """Return a canonical WorkingState payload for the runtime loop.""" working = deepcopy(hypothesis or {}) frames = _goal_frames_from_payload(working) if not frames: frames = [_compat_goal_frame(working)] raw_disclosure = _disclosure_policy_from_payload(working) protected_disclosure = _disclosure_policy_is_protected(raw_disclosure) static_contract_turn = protected_disclosure or _primary_frame_is_static_contract(frames) external_required = False if static_contract_turn else _requires_external_evidence(working=working, frames=frames) frames = [_normalize_frame(frame, external_required=external_required) for frame in frames] primary_frame = frames[0] if frames else {} gaps = _operation_gaps_from_payload(working) has_explicit_gap_payload = isinstance(working.get("operation_gaps"), list) if not gaps and not has_explicit_gap_payload: gaps = _compat_operation_gaps(working) if static_contract_turn: gaps = [gap for gap in gaps if str(gap.get("gap_type") or "") in USER_GAP_TYPES] gaps = _normalize_operation_gaps(gaps, primary_frame=primary_frame, external_required=external_required) if ( external_required and not _has_blocking_user_gap(working=working, gaps=gaps) and not any(_canonical_hint(gap) in EXTERNAL_READ_ACTIONS for gap in gaps) ): gaps.append( { "key": "fresh_external_evidence", "gap_type": "freshness_gap", "blocking": False, "user_actionable": False, "customer_question": "", "system_action_hint": "external_search", "reason": "需要读取外部公开新近证据。", } ) working["goal_frames"] = frames working["operation_gaps"] = gaps working["disclosure_policy"] = _normalize_disclosure_policy(raw_disclosure, primary_frame=primary_frame) working["freshness_required"] = bool(working.get("freshness_required")) or external_required working["contract_version"] = "pmflow_ai_turn_contract.v1" working["legacy_compat_consumed"] = bool(working.get("decision_type") or working.get("evidence_needs")) return working def _primary_frame_is_static_contract(frames: list[dict[str, Any]]) -> bool: if not frames: return False return str(frames[0].get("operation_kind") or "").strip() == "explain_assistant" def _disclosure_policy_from_payload(payload: dict[str, Any]) -> dict[str, Any]: raw = payload.get("disclosure_policy") return deepcopy(raw) if isinstance(raw, dict) else {} def _disclosure_policy_is_protected(raw: dict[str, Any]) -> bool: visibility = str(raw.get("visibility_class") or "").strip() response_mode = str(raw.get("allowed_response_mode") or "").strip() return visibility == "protected_pmflow_internal" or response_mode == "protected_refusal" def _normalize_disclosure_policy(raw: dict[str, Any], *, primary_frame: dict[str, Any]) -> dict[str, Any]: visibility = str(raw.get("visibility_class") or "").strip() if visibility not in DISCLOSURE_VISIBILITY_CLASSES: visibility = _default_visibility_for_frame(primary_frame) response_mode = str(raw.get("allowed_response_mode") or "").strip() if response_mode not in DISCLOSURE_RESPONSE_MODES: response_mode = _default_response_mode_for_visibility(visibility=visibility, primary_frame=primary_frame) categories = [ str(item or "").strip()[:80] for item in list(raw.get("protected_categories") or []) if str(item or "").strip() ][:12] contract_key = str(raw.get("contract_key") or "").strip()[:120] if visibility == "protected_pmflow_internal" or response_mode == "protected_refusal": visibility = "protected_pmflow_internal" response_mode = "protected_refusal" contract_key = contract_key or "protected_internal_refusal.v1" if not categories: categories = ["pmflow_internal_implementation"] elif str(primary_frame.get("operation_kind") or "").strip() == "explain_assistant": response_mode = "safe_product_boundary" contract_key = contract_key or "assistant_identity.v1" return { "visibility_class": visibility, "protected_categories": categories, "allowed_response_mode": response_mode, "policy_reason": str(raw.get("policy_reason") or "")[:400], "contract_key": contract_key, } def _default_visibility_for_frame(primary_frame: dict[str, Any]) -> DisclosureVisibilityClass: source = _source_policy_from_frame(primary_frame) if source == "external": return "external_public" if str(primary_frame.get("operation_kind") or "").strip() in {"list_collection", "inspect_object", "guide_execution"}: return "tenant_business_state" return "public_product_capability" def _default_response_mode_for_visibility(*, visibility: str, primary_frame: dict[str, Any]) -> DisclosureResponseMode: if visibility == "protected_pmflow_internal": return "protected_refusal" if visibility == "platform_admin_debug": return "platform_admin_redirect" if str(primary_frame.get("operation_kind") or "").strip() == "explain_assistant": return "safe_product_boundary" return "answer" def _has_blocking_user_gap(*, working: dict[str, Any], gaps: list[dict[str, Any]]) -> bool: if bool(working.get("can_answer_with_assumptions")): return False for gap in gaps: if not isinstance(gap, dict): continue if bool(gap.get("blocking")) and str(gap.get("gap_type") or "") in USER_GAP_TYPES: return True return False def canonical_action_hint_for_gap(*, hint: str, gap_type: str, frame: PMFlowAIGoalFrame | dict[str, Any] | None) -> str: """Canonicalize a system gap into a runtime action, or return empty.""" normalized = str(hint or "").strip() if normalized in READ_RUNTIME_ACTIONS: return normalized if str(gap_type or "").strip() == "freshness_gap": return "external_search" if normalized and _source_policy_from_frame(frame) == "external": return "external_search" if not normalized and _source_policy_from_frame(frame) == "external": return "external_search" return "" def validate_goal_frame_contract(frame: PMFlowAIGoalFrame | dict[str, Any]) -> list[str]: raw = _frame_dict(frame) operation = str(raw.get("operation_kind") or "answer_general") source = _source_policy_from_frame(raw) shape = str(raw.get("answer_shape") or "general_answer") issues: list[str] = [] if source not in SOURCE_SCOPE_BY_OPERATION.get(operation, ("internal", "external", "hybrid", "none")): issues.append(f"source_policy {source!r} is not allowed for operation {operation!r}") if shape not in ANSWER_SHAPES_BY_OPERATION.get(operation, ("general_answer",)): issues.append(f"answer_shape {shape!r} is not allowed for operation {operation!r}") collection = raw.get("target_collection") if operation == "list_collection" and not isinstance(collection, dict): issues.append("list_collection requires target_collection") if operation != "list_collection" and isinstance(collection, dict) and str(collection.get("object_type") or "") == "project": if source == "external": issues.append("external frame cannot use project collection evidence") return issues def validate_turn_state_contract(turn_state: PMFlowAITurnState) -> list[str]: issues: list[str] = [] if turn_state.selected_frame is not None: issues.extend(validate_goal_frame_contract(turn_state.selected_frame)) for gap in list(turn_state.operation_gaps or []): if gap.gap_type in USER_GAP_TYPES: if not gap.user_actionable: issues.append(f"user gap {gap.key!r} must be user_actionable") continue if gap.gap_type in SYSTEM_GAP_TYPES: if gap.user_actionable: issues.append(f"system gap {gap.key!r} cannot be user_actionable") hint = canonical_action_hint_for_gap( hint=str(gap.system_action_hint or ""), gap_type=str(gap.gap_type or ""), frame=turn_state.selected_frame, ) if not hint: issues.append(f"system gap {gap.key!r} has no canonical read action") return issues def _goal_frames_from_payload(payload: dict[str, Any]) -> list[dict[str, Any]]: rows = payload.get("goal_frames") return [deepcopy(item) for item in list(rows or []) if isinstance(item, dict)] def _operation_gaps_from_payload(payload: dict[str, Any]) -> list[dict[str, Any]]: rows = payload.get("operation_gaps") return [deepcopy(item) for item in list(rows or []) if isinstance(item, dict)] def _compat_goal_frame(payload: dict[str, Any]) -> dict[str, Any]: answer_shape = _coerce_answer_shape(payload.get("answer_shape")) primary = payload.get("primary_object") if isinstance(payload.get("primary_object"), dict) else {} binding = str(primary.get("binding_strength") or "").strip() has_primary = binding not in {"", "none", "context", "visible_context"} needs = {str(item or "").strip() for item in list(payload.get("evidence_needs") or []) if str(item or "").strip()} external_required = bool(payload.get("freshness_required")) or "external_fresh" in needs internal_required = bool(needs.intersection({"surface_context", "internal_projects", "saved_ideas", "project_lookup", "project_status"})) if answer_shape == "fact_list": operation: OperationKind = "list_collection" scope_type = "current_space" source_policy = "internal" target_collection = { "object_type": "saved_idea" if "saved_ideas" in needs else "project", "required": True, "scope": {"scope_type": scope_type, "binding_strength": "surface", "source": "legacy_compat"}, } elif (has_primary or "project_lookup" in needs or "project_status" in needs) and answer_shape in {"status_answer", "execution_guidance", "composite_answer"}: operation = "inspect_object" scope_type = "current_project" if has_primary else "current_space" source_policy = "internal" target_collection = None elif external_required or answer_shape == "strategic_judgement": operation = "judge_direction" scope_type = "active_workspace" if internal_required and external_required else "external_public" source_policy = "hybrid" if internal_required and external_required else "external" target_collection = None elif answer_shape in {"execution_guidance", "status_answer", "composite_answer"}: operation = "guide_execution" scope_type = "current_space" source_policy = "internal" target_collection = None else: operation = "answer_general" scope_type = "current_space" source_policy = "internal" target_collection = None frame = { "frame_id": "contract_compat_frame", "operation_kind": operation, "target_scope": { "scope_type": scope_type, "binding_strength": "explicit" if has_primary else "surface", "source": "legacy_compat", }, "target_collection": target_collection, "target_object_requirement": "single" if has_primary and operation == "inspect_object" else "none", "candidate_object": primary if has_primary else None, "source_policy": source_policy, "answer_shape": answer_shape, "confidence": 0.55, "assumptions": ["由旧兼容字段归一为 WorkingState 合同。"], "success_criteria": list(payload.get("success_criteria") or []), } return {key: value for key, value in frame.items() if value is not None} def _compat_operation_gaps(payload: dict[str, Any]) -> list[dict[str, Any]]: needs: list[str] = [] for item in list(payload.get("evidence_needs") or []): text = str(item or "").strip() if text and text not in needs: needs.append(text) if bool(payload.get("freshness_required")): if "external_fresh" not in needs: needs.append("external_fresh") if bool(payload.get("needs_user_input")) and not bool(payload.get("can_answer_with_assumptions")): if "clarification" not in needs: needs.append("clarification") out: list[dict[str, Any]] = [] for key in needs: if key == "synthesize": continue if key == "clarification": out.append( { "key": "user_clarification_required", "gap_type": "user_clarification_gap", "blocking": True, "user_actionable": True, "customer_question": "请补充继续判断必须由你确认的信息。", "system_action_hint": "", "reason": "需要用户补充阻塞信息。", } ) else: out.append( { "key": key, "gap_type": "freshness_gap" if key == "external_fresh" else "system_evidence_gap", "blocking": False, "user_actionable": False, "customer_question": "", "system_action_hint": _compat_action_for_need(key), "reason": "需要读取系统证据。", } ) return out def _normalize_frame(frame: dict[str, Any], *, external_required: bool) -> dict[str, Any]: out = deepcopy(frame) scope = out.get("target_scope") if isinstance(out.get("target_scope"), dict) else {} collection = out.get("target_collection") if isinstance(out.get("target_collection"), dict) else None operation = str(out.get("operation_kind") or "answer_general") answer_shape = _coerce_answer_shape(out.get("answer_shape")) source_policy = _source_policy_from_frame(out) if external_required and source_policy == "hybrid" and answer_shape in {"strategic_judgement", "composite_answer", "general_answer"} and not collection: operation = "judge_direction" scope = {"scope_type": "active_workspace", "binding_strength": "surface", "source": str(scope.get("source") or "contract_normalizer")} out["target_object_requirement"] = "none" out.pop("target_collection", None) elif external_required and answer_shape in {"strategic_judgement", "composite_answer", "general_answer"} and not collection: operation = "judge_direction" source_policy = "external" scope = {"scope_type": "external_public", "binding_strength": "surface", "source": str(scope.get("source") or "contract_normalizer")} out["target_object_requirement"] = "none" out.pop("target_collection", None) elif operation == "list_collection" and collection: source_policy = "internal" scope = scope or {"scope_type": "current_space", "binding_strength": "surface", "source": "contract_normalizer"} answer_shape = "fact_list" elif source_policy == "none": source_policy = _source_policy_from_scope(scope) out["operation_kind"] = operation out["answer_shape"] = answer_shape out["source_policy"] = source_policy out["target_scope"] = _normalize_scope(scope, source_policy=source_policy) if collection: collection.setdefault("required", True) collection.setdefault("scope", out["target_scope"]) out["target_collection"] = collection return out def _normalize_operation_gaps( gaps: list[dict[str, Any]], *, primary_frame: dict[str, Any], external_required: bool, ) -> list[dict[str, Any]]: out: list[dict[str, Any]] = [] for index, raw in enumerate(gaps[:20], start=1): gap = deepcopy(raw) gap_type = str(gap.get("gap_type") or "system_evidence_gap") gap["key"] = str(gap.get("key") or f"operation_gap_{index}")[:120] if gap_type in USER_GAP_TYPES: gap["user_actionable"] = True gap["system_action_hint"] = "" elif gap_type not in SYSTEM_GAP_TYPES: gap["user_actionable"] = False gap["customer_question"] = "" gap["system_action_hint"] = "" else: gap["user_actionable"] = False gap["customer_question"] = "" hint = canonical_action_hint_for_gap( hint=str(gap.get("system_action_hint") or ""), gap_type=gap_type, frame=primary_frame, ) if not hint and external_required: hint = "external_search" gap["system_action_hint"] = hint if external_required and hint == "external_search" and gap_type == "system_evidence_gap": gap["gap_type"] = "freshness_gap" out.append(gap) return out def _requires_external_evidence(*, working: dict[str, Any], frames: list[dict[str, Any]]) -> bool: if bool(working.get("freshness_required")): return True for frame in frames: if _source_policy_from_frame(frame) == "external": return True if str(frame.get("answer_shape") or "") == "strategic_judgement" and not isinstance(frame.get("target_collection"), dict): scope = frame.get("target_scope") if isinstance(frame.get("target_scope"), dict) else {} if str(scope.get("scope_type") or "") in {"external_public", "unspecified", ""}: return True for gap in _operation_gaps_from_payload(working): gap_type = str(gap.get("gap_type") or "") hint = str(gap.get("system_action_hint") or "").strip() if gap_type == "freshness_gap": return True if hint in EXTERNAL_READ_ACTIONS: return True needs = {str(item or "").strip() for item in list(working.get("evidence_needs") or []) if str(item or "").strip()} return "external_fresh" in needs def _compat_action_for_need(key: str) -> str: return { "surface_context": "inspect_surface_context", "internal_projects": "inspect_internal_projects", "saved_ideas": "inspect_saved_ideas", "project_lookup": "resolve_project_objects", "project_status": "inspect_project_status", "external_fresh": "external_search", }.get(key, "") def _source_policy_from_frame(frame: PMFlowAIGoalFrame | dict[str, Any] | None) -> SourcePolicy: raw = _frame_dict(frame) explicit = str(raw.get("source_policy") or "").strip() if explicit in {"internal", "external", "hybrid", "none"}: return explicit # type: ignore[return-value] scope = raw.get("target_scope") if isinstance(raw.get("target_scope"), dict) else {} return _source_policy_from_scope(scope) def _source_policy_from_scope(scope: dict[str, Any]) -> SourcePolicy: scope_type = str(scope.get("scope_type") or "").strip() if scope_type == "external_public": return "external" if scope_type in {"current_space", "current_project"}: return "internal" if scope_type == "active_workspace": return "hybrid" return "none" def _normalize_scope(scope: dict[str, Any], *, source_policy: str) -> dict[str, Any]: out = dict(scope or {}) if not str(out.get("scope_type") or "").strip() or str(out.get("scope_type")) == "unspecified": if source_policy == "external": out["scope_type"] = "external_public" elif source_policy == "internal": out["scope_type"] = "current_space" elif source_policy == "hybrid": out["scope_type"] = "active_workspace" else: out["scope_type"] = "unspecified" out.setdefault("binding_strength", "surface" if source_policy in {"internal", "external", "hybrid"} else "none") out.setdefault("source", "contract_normalizer") return out def _coerce_answer_shape(value: Any) -> AnswerShape: text = str(value or "").strip().lower() if text in { "strategic_judgement", "status_answer", "clarification", "execution_guidance", "fact_list", "composite_answer", "general_answer", }: return text # type: ignore[return-value] return "general_answer" def _canonical_hint(gap: dict[str, Any]) -> str: return str(gap.get("system_action_hint") or "").strip() def _frame_dict(frame: PMFlowAIGoalFrame | dict[str, Any] | None) -> dict[str, Any]: if isinstance(frame, PMFlowAIGoalFrame): return frame.model_dump(mode="json", exclude_none=True) return dict(frame or {}) if isinstance(frame, dict) else {} __all__ = [ "CANONICAL_RUNTIME_ACTIONS", "EXTERNAL_READ_ACTIONS", "INTERNAL_READ_ACTIONS", "READ_RUNTIME_ACTIONS", "canonical_action_hint_for_gap", "normalize_turn_hypothesis_contract", "validate_goal_frame_contract", "validate_turn_state_contract", ]