552 lines
24 KiB
Python
552 lines
24 KiB
Python
"""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",
|
|
]
|