"""WorkingState revision for the PMFlow AI turn runtime. This module does not classify raw language and does not render answers. It only turns structured observations and verification feedback into state patches so the runtime loop can keep probing, reading, revising, and verifying. """ from __future__ import annotations from typing import Any from ..schemas.pmflow_ai_turn_runtime import ( PMFlowAIDisclosureState, PMFlowAIEvidenceGap, PMFlowAIGoalFrame, PMFlowAIOperationGap, PMFlowAIObservation, PMFlowAIObservationItem, PMFlowAIStateRevision, PMFlowAITargetCollection, PMFlowAITargetScope, PMFlowAITurnObjectRef, PMFlowAITurnState, PMFlowAIVerificationFailure, ) from .pmflow_ai_turn_contract import normalize_turn_hypothesis_contract class PMFlowAITurnStateReviser: """Revise the WorkingState from observations and verification feedback.""" def initialize_working_state(self, *, turn_state: PMFlowAITurnState, hypothesis: dict[str, Any]) -> PMFlowAITurnState: hypothesis = normalize_turn_hypothesis_contract(hypothesis) frames = _frames_from_hypothesis(turn_state=turn_state, hypothesis=hypothesis) if frames and not turn_state.candidate_frames: turn_state.candidate_frames = frames if turn_state.candidate_frames and turn_state.selected_frame is None: turn_state.selected_frame = turn_state.candidate_frames[0] turn_state.disclosure_state = _disclosure_state_from_hypothesis(hypothesis) turn_state.operation_gaps = _operation_gaps_from_hypothesis(hypothesis) or _operation_gaps_from_evidence_gaps(turn_state.evidence_state.evidence_gaps) return turn_state def observe_payload( self, *, turn_state: PMFlowAITurnState, action_type: str, payload: dict[str, Any] | None, ) -> PMFlowAIObservation | None: observation = _observation_from_payload(turn_state=turn_state, action_type=action_type, payload=payload) if observation is not None: turn_state.observations.append(observation) return observation def revise_from_observation( self, *, turn_state: PMFlowAITurnState, observation: PMFlowAIObservation | None, ) -> PMFlowAITurnState: if observation is None: return turn_state patch: dict[str, Any] = {} selected = turn_state.selected_frame if selected is None: selected = _select_frame_for_observation(turn_state=turn_state, observation=observation) if selected is not None: turn_state.selected_frame = selected patch["selected_frame_id"] = selected.frame_id reframed = _reframe_from_collection_observation(turn_state=turn_state, observation=observation) if reframed is not None: selected = reframed patch["selected_frame_id"] = selected.frame_id patch["reframed_from_observation"] = True if selected and selected.operation_kind in observation.can_answer_operations: turn_state.sufficiency_state.can_answer_now = True turn_state.sufficiency_state.needs_user_input = False patch["can_answer_now"] = True patch["must_ask_user"] = False if observation.remaining_gaps: turn_state.operation_gaps = _merge_operation_gaps(turn_state.operation_gaps, observation.remaining_gaps) if observation.observation_type == "collection_snapshot" and selected and selected.operation_kind == "list_collection": turn_state.customer_packet_draft.update( { "operation_kind": "list_collection", "object_type": observation.object_type, "scope_type": observation.scope.scope_type, "items": [item.model_dump(mode="json", exclude_none=True) for item in observation.items[:20]], "count": observation.count, } ) turn_state.sufficiency_state.needs_user_input = False patch["customer_packet_draft"] = {"operation_kind": "list_collection", "object_type": observation.object_type} if patch: turn_state.state_revisions.append( PMFlowAIStateRevision( revision_id=f"rev_{len(turn_state.state_revisions) + 1}", source_observation_id=observation.observation_id, selected_frame_id=(turn_state.selected_frame.frame_id if turn_state.selected_frame else ""), reason="Observation revised WorkingState.", state_patch=patch, ) ) return turn_state def revise_from_verification( self, *, turn_state: PMFlowAITurnState, failure: PMFlowAIVerificationFailure | None, ) -> PMFlowAITurnState: if failure is None: return turn_state turn_state.verification_failures.append(failure) patch = dict(failure.state_patch or {}) if patch.get("must_ask_user") is False: turn_state.sufficiency_state.needs_user_input = False if patch.get("can_answer_now") is True: turn_state.sufficiency_state.can_answer_now = True if patch.get("can_answer_now") is False: turn_state.sufficiency_state.can_answer_now = False turn_state.completed = False if patch.get("needs_internal_evidence") is True: turn_state.sufficiency_state.needs_internal_evidence = True if patch.get("needs_external_evidence") is True: turn_state.sufficiency_state.needs_external_evidence = True if isinstance(patch.get("disclosure_state"), dict): turn_state.disclosure_state = _disclosure_state_from_hypothesis({"disclosure_policy": patch.get("disclosure_state")}) next_action = str(failure.next_action_hint or "").strip() customer_reason = str(failure.customer_safe_summary or "").strip() if not customer_reason: customer_reason = "这轮内部校验没有形成可安全展示的结论，需要先补齐可验证依据或改用安全边界回答。" if ( next_action and bool(failure.retry_would_change_state) and next_action not in {"render_or_fail_closed", "select_minimal_read_or_render_action"} ): turn_state.operation_gaps = _merge_operation_gaps( turn_state.operation_gaps, [ PMFlowAIOperationGap( key=f"verification_{next_action}", gap_type="system_evidence_gap", blocking=True, user_actionable=False, system_action_hint=next_action, reason=customer_reason, ) ], ) turn_state.state_revisions.append( PMFlowAIStateRevision( revision_id=f"rev_{len(turn_state.state_revisions) + 1}", reason=customer_reason, selected_frame_id=(turn_state.selected_frame.frame_id if turn_state.selected_frame else ""), state_patch=patch, ) ) return turn_state def _frames_from_hypothesis(*, turn_state: PMFlowAITurnState, hypothesis: dict[str, Any]) -> list[PMFlowAIGoalFrame]: direct_frames = _frames_from_goal_frame_payload(turn_state=turn_state, hypothesis=hypothesis) if direct_frames: return direct_frames return [ PMFlowAIGoalFrame( frame_id="fallback_general_frame", operation_kind="answer_general", target_scope=PMFlowAITargetScope(scope_type="current_space", binding_strength="surface", source="contract_fallback"), target_object_requirement="none", answer_shape=turn_state.answer_state.answer_shape, source_policy="internal", confidence=0.2, assumptions=["缺少直接 GoalFrame，按通用回答合同降级。"], success_criteria=list(turn_state.turn_goal.success_criteria or []), ) ] def _disclosure_state_from_hypothesis(hypothesis: dict[str, Any]) -> PMFlowAIDisclosureState: raw = hypothesis.get("disclosure_policy") if isinstance(hypothesis.get("disclosure_policy"), dict) else {} try: return PMFlowAIDisclosureState.model_validate(raw) except Exception: return PMFlowAIDisclosureState() def _frames_from_goal_frame_payload(*, turn_state: PMFlowAITurnState, hypothesis: dict[str, Any]) -> list[PMFlowAIGoalFrame]: raw_frames = hypothesis.get("goal_frames") rows = raw_frames if isinstance(raw_frames, list) else [] frames: list[PMFlowAIGoalFrame] = [] for index, raw in enumerate(rows[:6], start=1): if not isinstance(raw, dict): continue scope = _target_scope_from_payload(raw.get("target_scope")) collection = None raw_collection = raw.get("target_collection") if isinstance(raw_collection, dict): collection = PMFlowAITargetCollection( object_type=str(raw_collection.get("object_type") or "")[:80], required=bool(raw_collection.get("required")), scope=_target_scope_from_payload(raw_collection.get("scope")) if isinstance(raw_collection.get("scope"), dict) else scope, ) candidate_object = None raw_candidate = raw.get("candidate_object") if isinstance(raw_candidate, dict): try: candidate_object = PMFlowAITurnObjectRef.model_validate(raw_candidate) except Exception: candidate_object = turn_state.object_state.primary_object try: frames.append( PMFlowAIGoalFrame( frame_id=str(raw.get("frame_id") or f"goal_frame_{index}")[:80], operation_kind=str(raw.get("operation_kind") or "answer_general"), target_scope=scope, target_collection=collection, target_object_requirement=str(raw.get("target_object_requirement") or "none"), candidate_object=candidate_object, source_policy=str(raw.get("source_policy") or "none"), answer_shape=str(raw.get("answer_shape") or turn_state.answer_state.answer_shape), confidence=float(raw.get("confidence") or 0.0), assumptions=list(raw.get("assumptions") or []), success_criteria=list(raw.get("success_criteria") or turn_state.turn_goal.success_criteria or []), ) ) except Exception: continue return frames def _reframe_from_collection_observation( *, turn_state: PMFlowAITurnState, observation: PMFlowAIObservation, ) -> PMFlowAIGoalFrame | None: selected = turn_state.selected_frame if observation.observation_type != "collection_snapshot": return None if not observation.items and observation.count in {None, 0}: return None if "list_collection" not in observation.can_answer_operations: return None if selected is not None and selected.operation_kind == "list_collection": return None if str(turn_state.answer_state.answer_shape or "") != "fact_list": return None scope = observation.scope or PMFlowAITargetScope(scope_type="current_space", binding_strength="surface", source="observation_reframe") frame = PMFlowAIGoalFrame( frame_id=f"observation_collection_{len(turn_state.candidate_frames) + 1}", operation_kind="list_collection", target_scope=scope, target_collection=PMFlowAITargetCollection( object_type=str(observation.object_type or "object")[:80], required=True, scope=scope, ), target_object_requirement="none", answer_shape="fact_list", source_policy="internal", confidence=0.78, assumptions=["根据已完成的集合观察修正本轮目标框架。"], success_criteria=list(turn_state.turn_goal.success_criteria or []), ) turn_state.candidate_frames.insert(0, frame) turn_state.selected_frame = frame turn_state.answer_state.answer_shape = "fact_list" turn_state.answer_state.required_sections = ["匹配结果", "依据", "下一步"] return frame def _target_scope_from_payload(value: Any) -> PMFlowAITargetScope: raw = value if isinstance(value, dict) else {} try: return PMFlowAITargetScope( scope_type=str(raw.get("scope_type") or "unspecified"), binding_strength=str(raw.get("binding_strength") or "none"), source=str(raw.get("source") or "goal_frame"), ) except Exception: return PMFlowAITargetScope() def _operation_gaps_from_hypothesis(hypothesis: dict[str, Any]) -> list[PMFlowAIOperationGap]: raw_gaps = hypothesis.get("operation_gaps") rows = raw_gaps if isinstance(raw_gaps, list) else [] gaps: list[PMFlowAIOperationGap] = [] for index, raw in enumerate(rows[:20], start=1): if not isinstance(raw, dict): continue try: gaps.append( PMFlowAIOperationGap( key=str(raw.get("key") or f"operation_gap_{index}")[:120], gap_type=str(raw.get("gap_type") or "system_evidence_gap"), blocking=bool(raw.get("blocking")), user_actionable=bool(raw.get("user_actionable")), customer_question=str(raw.get("customer_question") or "")[:240], system_action_hint=str(raw.get("system_action_hint") or "")[:160], reason=str(raw.get("reason") or raw.get("label") or "")[:400], ) ) except Exception: continue return gaps def _operation_gaps_from_evidence_gaps(gaps: list[PMFlowAIEvidenceGap]) -> list[PMFlowAIOperationGap]: out: list[PMFlowAIOperationGap] = [] for gap in list(gaps or [])[:20]: out.append( PMFlowAIOperationGap( key=gap.key, gap_type=gap.gap_type, blocking=gap.severity == "blocking", user_actionable=gap.user_actionable, customer_question=gap.customer_question, system_action_hint=gap.system_action_hint, reason=gap.reason or gap.label, ) ) return out def _observation_from_payload( *, turn_state: PMFlowAITurnState, action_type: str, payload: dict[str, Any] | None, ) -> PMFlowAIObservation | None: if not isinstance(payload, dict): return PMFlowAIObservation( observation_id=f"obs_{len(turn_state.observations) + 1}", observation_type="verification_feedback", action_phase="verify", confidence="unknown", remaining_gaps=[ PMFlowAIOperationGap( key=f"{action_type}_no_payload", gap_type="system_evidence_gap", blocking=False, system_action_hint="select_next_action", reason="Action returned no payload.", ) ], ) raw_observation = payload.get("runtime_observation") if isinstance(raw_observation, dict): try: observation = PMFlowAIObservation.model_validate(raw_observation) if not observation.observation_id: observation.observation_id = f"obs_{len(turn_state.observations) + 1}" return observation except Exception: pass tool_result = payload.get("runtime_evidence_tool_result") if isinstance(payload.get("runtime_evidence_tool_result"), dict) else {} findings = [item for item in list(tool_result.get("findings") or []) if isinstance(item, dict)] items = [ PMFlowAIObservationItem( object_type=_object_type_from_action(action_type), title=str(item.get("fact") or "")[:200], summary=str(item.get("implication") or item.get("fact") or "")[:400], ) for item in findings[:20] ] can_answer = _can_answer_operations_from_action(action_type=action_type, turn_state=turn_state, payload=payload) confidence = str(tool_result.get("confidence") or ("high" if payload.get("_runtime_payload_kind") == "final_answer" else "unknown")) return PMFlowAIObservation( observation_id=f"obs_{len(turn_state.observations) + 1}", observation_type=_observation_type_from_action(action_type=action_type, payload=payload), action_phase="read" if action_type != "synthesize_answer" else "draft", scope=(turn_state.selected_frame.target_scope if turn_state.selected_frame else PMFlowAITargetScope()), object_type=_object_type_from_action(action_type), items=items, count=len(items) if items else None, can_answer_operations=can_answer, cannot_answer_operations=[], confidence=confidence, remaining_gaps=_operation_gaps_from_raw(tool_result.get("remaining_gaps")), summary=str(tool_result.get("summary") or payload.get("answer") or payload.get("reply_text") or "")[:500], ) def _observation_type_from_action(*, action_type: str, payload: dict[str, Any]) -> str: if payload.get("_runtime_payload_kind") == "final_answer": return "verification_feedback" if action_type == "inspect_surface_context": return "collection_snapshot" if action_type in {"resolve_project_objects", "inspect_project_status", "inspect_internal_projects"}: return "object_snapshot" if action_type == "inspect_saved_ideas": return "collection_snapshot" if action_type in {"external_search", "external_fetch"}: return "external_evidence" return "scope_probe" def _object_type_from_action(action_type: str) -> str: if action_type == "inspect_saved_ideas": return "saved_idea" if action_type in {"inspect_surface_context", "resolve_project_objects", "inspect_project_status", "inspect_internal_projects"}: return "project" if action_type in {"external_search", "external_fetch"}: return "external_source" return "object" def _can_answer_operations_from_action(*, action_type: str, turn_state: PMFlowAITurnState, payload: dict[str, Any]) -> list[str]: selected = turn_state.selected_frame.operation_kind if turn_state.selected_frame else "answer_general" if payload.get("_runtime_payload_kind") == "final_answer": return [selected] if action_type in {"inspect_surface_context", "inspect_saved_ideas", "resolve_project_objects"}: return ["list_collection"] if selected == "list_collection" else [] if action_type in {"inspect_project_status", "inspect_internal_projects"}: return ["inspect_object"] if selected == "inspect_object" else [] if action_type in {"external_search", "external_fetch"}: return ["judge_direction", "compare_objects", "answer_general"] return [] def _operation_gaps_from_raw(values: Any) -> list[PMFlowAIOperationGap]: rows = values if isinstance(values, list) else [] out: list[PMFlowAIOperationGap] = [] for index, value in enumerate(rows[:10], start=1): text = str(value or "").strip() if not text: continue out.append( PMFlowAIOperationGap( key=f"tool_gap_{index}", gap_type="system_evidence_gap", blocking=False, system_action_hint="select_next_action", reason=text, ) ) return out def _select_frame_for_observation(*, turn_state: PMFlowAITurnState, observation: PMFlowAIObservation) -> PMFlowAIGoalFrame | None: for frame in list(turn_state.candidate_frames or []): if frame.operation_kind in observation.can_answer_operations: return frame return turn_state.candidate_frames[0] if turn_state.candidate_frames else None def _merge_operation_gaps(existing: list[PMFlowAIOperationGap], incoming: list[PMFlowAIOperationGap]) -> list[PMFlowAIOperationGap]: by_key = {gap.key: gap for gap in list(existing or [])} for gap in incoming: by_key[gap.key] = gap return list(by_key.values())[:20] __all__ = ["PMFlowAITurnStateReviser"]