"""Contracts for the bounded PMFlow AI turn runtime. These models are customer-safe runtime snapshots. They describe how one user turn was understood, what evidence was inspected, which bounded actions ran, and why the runtime answered, asked, or deferred to confirmation. """ from __future__ import annotations from typing import Any, Literal from pydantic import BaseModel, Field, field_validator, model_validator PMFLOW_AI_TURN_STATE_VERSION = "pmflow_ai_turn_state.v1" RuntimeActionKind = Literal[ "inspect_surface_context", "inspect_internal_projects", "inspect_saved_ideas", "resolve_project_objects", "inspect_project_status", "external_search", "external_fetch", "synthesize_answer", "ask_clarification", ] RuntimeActionStatus = Literal["candidate", "performed", "blocked", "skipped"] RuntimeStepStatus = Literal["pending", "active", "completed", "skipped"] RuntimeActionPhase = Literal["probe", "read", "draft", "prepare", "verify"] SourcePolicy = Literal["internal", "external", "hybrid", "none"] OperationKind = Literal[ "explain_assistant", "list_collection", "inspect_object", "compare_objects", "judge_direction", "guide_execution", "answer_general", ] TargetScopeType = Literal[ "current_space", "current_project", "active_workspace", "external_public", "unspecified", ] TargetScopeBindingStrength = Literal["explicit", "surface", "workspace", "none"] TargetObjectRequirement = Literal["none", "single", "multiple", "authorization"] GapType = Literal[ "system_evidence_gap", "scope_gap", "object_binding_gap", "user_clarification_gap", "user_authorization_gap", "business_assumption_gap", "freshness_gap", ] ObservationType = Literal[ "scope_probe", "focus_probe", "collection_snapshot", "object_snapshot", "assistant_profile", "workspace_continuity", "write_or_execute_risk", "external_evidence", "verification_feedback", ] AnswerShape = Literal[ "strategic_judgement", "status_answer", "clarification", "execution_guidance", "fact_list", "composite_answer", "general_answer", ] EvidenceConfidence = Literal["high", "medium", "low", "unknown"] ProblemDecisionType = Literal[ "market_opportunity", "project_status", "execution_guidance", "object_lookup", "general_answer", ] ThreadWorkspaceKind = Literal[ "new_project_ideation", "external_research", "hybrid_research", "internal_research", "project_status_review", "execution_guidance", ] DisclosureVisibilityClass = Literal[ "public_product_capability", "tenant_business_state", "external_public", "protected_pmflow_internal", "platform_admin_debug", ] DisclosureResponseMode = Literal[ "answer", "safe_product_boundary", "protected_refusal", "platform_admin_redirect", ] class PMFlowAITargetScope(BaseModel): scope_type: TargetScopeType = "unspecified" binding_strength: TargetScopeBindingStrength = "none" source: str = Field(default="", max_length=120) @field_validator("source", mode="before") @classmethod def _sanitize_text(cls, value: Any) -> str: return _clean_text(value, limit=120) class PMFlowAITargetCollection(BaseModel): object_type: str = Field(default="", max_length=80) required: bool = False scope: PMFlowAITargetScope = Field(default_factory=PMFlowAITargetScope) @field_validator("object_type", mode="before") @classmethod def _sanitize_text(cls, value: Any) -> str: return _clean_text(value, limit=80) class PMFlowAIGoalFrame(BaseModel): frame_id: str = Field(default="", max_length=80) operation_kind: OperationKind = "answer_general" target_scope: PMFlowAITargetScope = Field(default_factory=PMFlowAITargetScope) target_collection: PMFlowAITargetCollection | None = None target_object_requirement: TargetObjectRequirement = "none" candidate_object: "PMFlowAITurnObjectRef | None" = None source_policy: SourcePolicy = "none" answer_shape: AnswerShape = "general_answer" confidence: float = Field(default=0.0, ge=0.0, le=1.0) assumptions: list[str] = Field(default_factory=list) success_criteria: list[str] = Field(default_factory=list) @field_validator("frame_id", mode="before") @classmethod def _sanitize_text(cls, value: Any) -> str: return _clean_text(value, limit=80) @field_validator("assumptions", "success_criteria", mode="before") @classmethod def _sanitize_list(cls, value: Any) -> list[str]: rows = value if isinstance(value, list) else [] return [_clean_text(item, limit=180) for item in rows[:10] if _clean_text(item, limit=180)] class PMFlowAIObservationItem(BaseModel): object_type: str = Field(default="", max_length=80) object_ref: str = Field(default="", max_length=160) title: str = Field(default="", max_length=200) status: str = Field(default="", max_length=120) href: str = Field(default="", max_length=240) summary: str = Field(default="", max_length=400) business_values: dict[str, Any] = Field(default_factory=dict) @field_validator("object_type", "object_ref", "title", "status", "href", "summary", mode="before") @classmethod def _sanitize_text(cls, value: Any) -> str: return _clean_text(value, limit=400) class PMFlowAIObservation(BaseModel): packet_version: str = "pmflow_ai_observation.v1" observation_id: str = Field(default="", max_length=80) observation_type: ObservationType action_phase: RuntimeActionPhase = "probe" scope: PMFlowAITargetScope = Field(default_factory=PMFlowAITargetScope) object_type: str = Field(default="", max_length=80) object_ref: str = Field(default="", max_length=160) items: list[PMFlowAIObservationItem] = Field(default_factory=list) count: int | None = Field(default=None, ge=0) can_answer_operations: list[OperationKind] = Field(default_factory=list) cannot_answer_operations: list[OperationKind] = Field(default_factory=list) confidence: EvidenceConfidence = "unknown" remaining_gaps: list["PMFlowAIOperationGap"] = Field(default_factory=list) summary: str = Field(default="", max_length=500) @field_validator("observation_id", "object_type", "object_ref", "summary", mode="before") @classmethod def _sanitize_text(cls, value: Any) -> str: return _clean_text(value, limit=500) @model_validator(mode="after") def _cap_items(self) -> "PMFlowAIObservation": self.items = list(self.items or [])[:20] self.remaining_gaps = list(self.remaining_gaps or [])[:12] return self class PMFlowAIOperationGap(BaseModel): key: str = Field(..., min_length=1, max_length=120) gap_type: GapType = "system_evidence_gap" blocking: bool = False user_actionable: bool = False customer_question: str = Field(default="", max_length=240) system_action_hint: str = Field(default="", max_length=160) reason: str = Field(default="", max_length=400) @field_validator("key", "customer_question", "system_action_hint", "reason", mode="before") @classmethod def _sanitize_text(cls, value: Any) -> str: return _clean_text(value, limit=400) class PMFlowAIStateRevision(BaseModel): revision_id: str = Field(default="", max_length=80) reason: str = Field(default="", max_length=400) source_observation_id: str = Field(default="", max_length=80) selected_frame_id: str = Field(default="", max_length=80) state_patch: dict[str, Any] = Field(default_factory=dict) @field_validator("revision_id", "reason", "source_observation_id", "selected_frame_id", mode="before") @classmethod def _sanitize_text(cls, value: Any) -> str: return _clean_text(value, limit=400) class PMFlowAIVerificationFailure(BaseModel): failure_id: str = Field(default="", max_length=80) failure_type: str = Field(default="", max_length=120) reason: str = Field(default="", max_length=400) internal_diagnostic: str = Field(default="", max_length=800) customer_safe_summary: str = Field(default="", max_length=400) customer_safe_effect: dict[str, Any] = Field(default_factory=dict) verification_delta: dict[str, Any] = Field(default_factory=dict) retry_would_change_state: bool = True state_patch: dict[str, Any] = Field(default_factory=dict) next_action_hint: str = Field(default="", max_length=120) @field_validator("failure_id", "failure_type", "reason", "customer_safe_summary", "next_action_hint", mode="before") @classmethod def _sanitize_text(cls, value: Any) -> str: return _clean_text(value, limit=400) @field_validator("internal_diagnostic", mode="before") @classmethod def _sanitize_internal_diagnostic(cls, value: Any) -> str: return _clean_text(value, limit=800) class PMFlowAIClaim(BaseModel): claim_id: str = Field(default="", max_length=80) claim_text: str = Field(default="", max_length=600) claim_type: str = Field(default="statement", max_length=80) support: list[str] = Field(default_factory=list) confidence: EvidenceConfidence = "unknown" customer_value: str = Field(default="", max_length=400) allowed_to_state: bool = True @field_validator("claim_id", "claim_type", mode="before") @classmethod def _sanitize_text(cls, value: Any) -> str: return _clean_text(value, limit=80) @field_validator("claim_text", mode="before") @classmethod def _sanitize_claim_text(cls, value: Any) -> str: return _clean_text(value, limit=600) @field_validator("customer_value", mode="before") @classmethod def _sanitize_customer_value(cls, value: Any) -> str: return _clean_text(value, limit=400) @field_validator("support", mode="before") @classmethod def _sanitize_support(cls, value: Any) -> list[str]: rows = value if isinstance(value, list) else [] return [_clean_text(item, limit=200) for item in rows[:8] if _clean_text(item, limit=200)] class PMFlowAIBoundary(BaseModel): boundary_id: str = Field(default="", max_length=80) blocked_claim: str = Field(default="", max_length=600) boundary_type: str = Field(default="", max_length=100) customer_safe_reason: str = Field(default="", max_length=500) allowed_alternative: str = Field(default="", max_length=500) support: list[str] = Field(default_factory=list) internal_diagnostic: str = Field(default="", max_length=800) @field_validator("boundary_id", "boundary_type", mode="before") @classmethod def _sanitize_text(cls, value: Any) -> str: return _clean_text(value, limit=100) @field_validator("blocked_claim", mode="before") @classmethod def _sanitize_blocked_claim(cls, value: Any) -> str: return _clean_text(value, limit=600) @field_validator("customer_safe_reason", "allowed_alternative", mode="before") @classmethod def _sanitize_customer_text(cls, value: Any) -> str: return _clean_text(value, limit=500) @field_validator("internal_diagnostic", mode="before") @classmethod def _sanitize_internal_diagnostic(cls, value: Any) -> str: return _clean_text(value, limit=800) @field_validator("support", mode="before") @classmethod def _sanitize_support(cls, value: Any) -> list[str]: rows = value if isinstance(value, list) else [] return [_clean_text(item, limit=200) for item in rows[:8] if _clean_text(item, limit=200)] class PMFlowAIAttempt(BaseModel): attempt_id: str = Field(default="", max_length=80) action_type: str = Field(default="", max_length=100) status: str = Field(default="", max_length=80) target: str = Field(default="", max_length=200) customer_visible_summary: str = Field(default="", max_length=500) produced_claim_ids: list[str] = Field(default_factory=list) produced_boundary_ids: list[str] = Field(default_factory=list) internal_diagnostic: str = Field(default="", max_length=800) @field_validator("attempt_id", "action_type", "status", mode="before") @classmethod def _sanitize_text(cls, value: Any) -> str: return _clean_text(value, limit=100) @field_validator("target", mode="before") @classmethod def _sanitize_target(cls, value: Any) -> str: return _clean_text(value, limit=200) @field_validator("customer_visible_summary", mode="before") @classmethod def _sanitize_summary(cls, value: Any) -> str: return _clean_text(value, limit=500) @field_validator("internal_diagnostic", mode="before") @classmethod def _sanitize_internal_diagnostic(cls, value: Any) -> str: return _clean_text(value, limit=800) @field_validator("produced_claim_ids", "produced_boundary_ids", mode="before") @classmethod def _sanitize_ids(cls, value: Any) -> list[str]: rows = value if isinstance(value, list) else [] return [_clean_text(item, limit=80) for item in rows[:12] if _clean_text(item, limit=80)] class PMFlowAIAnswerIntentPlan(BaseModel): intent_id: str = Field(default="answer_intent", max_length=80) opening_move: str = Field(default="direct_answer", max_length=120) answer_focus: str = Field(default="", max_length=400) user_value_focus: str = Field(default="", max_length=400) detail_level: str = Field(default="auto", max_length=80) structure_preference: str = Field(default="natural", max_length=120) lead_claim_id: str = Field(default="", max_length=80) lead_boundary_id: str = Field(default="", max_length=80) answer_moves: list[str] = Field(default_factory=list) composition_guidance: list[str] = Field(default_factory=list) anti_template_rules: list[str] = Field(default_factory=list) avoid_sections: bool = False avoid_repeating_previous: bool = True must_include_claim_ids: list[str] = Field(default_factory=list) must_include_boundary_ids: list[str] = Field(default_factory=list) optional_claim_ids: list[str] = Field(default_factory=list) ask_user: bool = False @field_validator("intent_id", "detail_level", mode="before") @classmethod def _sanitize_text(cls, value: Any) -> str: return _clean_text(value, limit=80) @field_validator("opening_move", "structure_preference", mode="before") @classmethod def _sanitize_short_text(cls, value: Any) -> str: return _clean_text(value, limit=120) @field_validator("answer_focus", mode="before") @classmethod def _sanitize_answer_focus(cls, value: Any) -> str: return _clean_text(value, limit=400) @field_validator("user_value_focus", mode="before") @classmethod def _sanitize_user_value_focus(cls, value: Any) -> str: return _clean_text(value, limit=400) @field_validator("lead_claim_id", "lead_boundary_id", mode="before") @classmethod def _sanitize_lead_ids(cls, value: Any) -> str: return _clean_text(value, limit=80) @field_validator("must_include_claim_ids", "must_include_boundary_ids", "optional_claim_ids", mode="before") @classmethod def _sanitize_ids(cls, value: Any) -> list[str]: rows = value if isinstance(value, list) else [] return [_clean_text(item, limit=80) for item in rows[:12] if _clean_text(item, limit=80)] @field_validator("answer_moves", "composition_guidance", "anti_template_rules", mode="before") @classmethod def _sanitize_rules(cls, value: Any) -> list[str]: rows = value if isinstance(value, list) else [] return [_clean_text(item, limit=220) for item in rows[:12] if _clean_text(item, limit=220)] class PMFlowAITerminalAnswerState(BaseModel): packet_version: str = "pmflow_ai_terminal_answer_state.v1" user_goal: str = Field(default="", max_length=400) target_object: str = Field(default="", max_length=240) terminal_status: str = Field(default="answerable", max_length=80) claims: list[PMFlowAIClaim] = Field(default_factory=list) boundaries: list[PMFlowAIBoundary] = Field(default_factory=list) attempts: list[PMFlowAIAttempt] = Field(default_factory=list) answer_intent_plan: PMFlowAIAnswerIntentPlan = Field(default_factory=PMFlowAIAnswerIntentPlan) remaining_gaps: list[str] = Field(default_factory=list) next_possible_actions: list[str] = Field(default_factory=list) confidence: EvidenceConfidence = "unknown" @field_validator("user_goal", "target_object", "terminal_status", mode="before") @classmethod def _sanitize_text(cls, value: Any) -> str: return _clean_text(value, limit=400) @field_validator("remaining_gaps", "next_possible_actions", mode="before") @classmethod def _sanitize_list(cls, value: Any) -> list[str]: rows = value if isinstance(value, list) else [] return [_clean_text(item, limit=240) for item in rows[:12] if _clean_text(item, limit=240)] @model_validator(mode="after") def _cap_lists(self) -> "PMFlowAITerminalAnswerState": self.claims = list(self.claims or [])[:20] self.boundaries = list(self.boundaries or [])[:20] self.attempts = list(self.attempts or [])[:20] return self def _clean_text(value: Any, *, limit: int = 500) -> str: return " ".join(str(value or "").strip().split())[:limit] class PMFlowAITurnGoal(BaseModel): user_goal: str = Field(default="answer_open_question", max_length=120) desired_outcome: str = Field(default="useful_answer", max_length=120) user_posture: str = Field(default="ask", max_length=80) success_criteria: list[str] = Field(default_factory=list) @field_validator("user_goal", "desired_outcome", "user_posture", mode="before") @classmethod def _sanitize_text(cls, value: Any) -> str: return _clean_text(value, limit=120) @field_validator("success_criteria", mode="before") @classmethod def _sanitize_list(cls, value: Any) -> list[str]: rows = value if isinstance(value, list) else [] return [_clean_text(item, limit=160) for item in rows[:8] if _clean_text(item, limit=160)] class PMFlowAITurnObjectRef(BaseModel): object_type: str = Field(default="unknown", max_length=80) object_ref: str = Field(default="", max_length=160) title: str = Field(default="", max_length=200) binding_strength: str = Field(default="none", max_length=80) role: str | None = Field(default=None, max_length=80) @field_validator("object_type", "object_ref", "title", "binding_strength", "role", mode="before") @classmethod def _sanitize_text(cls, value: Any) -> str | None: if value is None: return None return _clean_text(value, limit=200) class PMFlowAITurnObjectState(BaseModel): primary_object: PMFlowAITurnObjectRef | None = None context_objects: list[PMFlowAITurnObjectRef] = Field(default_factory=list) forbidden_substitution: bool = False workspace_focus: dict[str, Any] = Field(default_factory=dict) class PMFlowAIProblemFrame(BaseModel): subject: str = Field(default="", max_length=240) subject_terms: list[str] = Field(default_factory=list) domain_hints: list[str] = Field(default_factory=list) ambiguity_points: list[str] = Field(default_factory=list) user_question_type: str = Field(default="", max_length=120) decision_type: ProblemDecisionType = "general_answer" @field_validator("subject", "user_question_type", mode="before") @classmethod def _sanitize_text(cls, value: Any) -> str: return _clean_text(value, limit=240) @field_validator("subject_terms", "domain_hints", "ambiguity_points", mode="before") @classmethod def _sanitize_list(cls, value: Any) -> list[str]: rows = value if isinstance(value, list) else [] return [_clean_text(item, limit=120) for item in rows[:12] if _clean_text(item, limit=120)] class PMFlowAIInterpretationBranch(BaseModel): branch_id: str = Field(..., min_length=1, max_length=80) name: str = Field(..., min_length=1, max_length=160) assumptions: list[str] = Field(default_factory=list) evidence_needs: list[str] = Field(default_factory=list) risk_hints: list[str] = Field(default_factory=list) confidence: EvidenceConfidence = "medium" @field_validator("branch_id", "name", mode="before") @classmethod def _sanitize_text(cls, value: Any) -> str: return _clean_text(value, limit=160) @field_validator("assumptions", "evidence_needs", "risk_hints", mode="before") @classmethod def _sanitize_list(cls, value: Any) -> list[str]: rows = value if isinstance(value, list) else [] return [_clean_text(item, limit=180) for item in rows[:10] if _clean_text(item, limit=180)] class PMFlowAIInterpretationState(BaseModel): branches: list[PMFlowAIInterpretationBranch] = Field(default_factory=list) selected_branch_id: str = Field(default="", max_length=80) compare_branches: bool = False assumption_policy: Literal["state_assumptions", "ask_if_blocking", "single_branch"] = "single_branch" @field_validator("selected_branch_id", mode="before") @classmethod def _sanitize_text(cls, value: Any) -> str: return _clean_text(value, limit=80) class PMFlowAIJudgementState(BaseModel): direct_answer: str = Field(default="", max_length=500) opportunity: str = Field(default="", max_length=500) can_do: list[str] = Field(default_factory=list) should_not_do: list[str] = Field(default_factory=list) better_angle: str = Field(default="", max_length=500) mvp: list[str] = Field(default_factory=list) risks: list[str] = Field(default_factory=list) validation_questions: list[str] = Field(default_factory=list) next_actions: list[str] = Field(default_factory=list) confidence: EvidenceConfidence = "unknown" @field_validator("direct_answer", "opportunity", "better_angle", mode="before") @classmethod def _sanitize_text(cls, value: Any) -> str: return _clean_text(value, limit=500) @field_validator("can_do", "should_not_do", "mvp", "risks", "validation_questions", "next_actions", mode="before") @classmethod def _sanitize_list(cls, value: Any) -> list[str]: rows = value if isinstance(value, list) else [] return [_clean_text(item, limit=220) for item in rows[:10] if _clean_text(item, limit=220)] class PMFlowAITurnDeltaAnswer(BaseModel): question_id: str = Field(default="", max_length=120) question_label: str = Field(default="", max_length=240) values: list[str] = Field(default_factory=list) labels: list[str] = Field(default_factory=list) display_value: str = Field(default="", max_length=400) @field_validator("question_id", "question_label", "display_value", mode="before") @classmethod def _sanitize_text(cls, value: Any) -> str: return _clean_text(value, limit=400) @field_validator("values", "labels", mode="before") @classmethod def _sanitize_list(cls, value: Any) -> list[str]: rows = value if isinstance(value, list) else [] return [_clean_text(item, limit=240) for item in rows[:8] if _clean_text(item, limit=240)] class PMFlowAITurnDelta(BaseModel): packet_version: str = "pmflow_ai_turn_delta.v1" source: Literal["user_message", "interaction_submission"] = "user_message" raw_question: str = Field(default="", max_length=1000) effective_question: str = Field(default="", max_length=1000) answers: list[PMFlowAITurnDeltaAnswer] = Field(default_factory=list) supplement_text: str = Field(default="", max_length=1200) @field_validator("raw_question", "effective_question", "supplement_text", mode="before") @classmethod def _sanitize_text(cls, value: Any) -> str: return _clean_text(value, limit=1200) class PMFlowAIThreadWorkspace(BaseModel): packet_version: str = "pmflow_ai_thread_workspace.v1" workspace_id: str = Field(default="", max_length=120) workspace_kind: ThreadWorkspaceKind = "external_research" topic: str = Field(default="", max_length=240) primary_object: PMFlowAITurnObjectRef | None = None active_goal: str = Field(default="", max_length=160) collected_inputs: dict[str, Any] = Field(default_factory=dict) pending_questions: list[str] = Field(default_factory=list) evidence_requirements: list[str] = Field(default_factory=list) forbidden_substitution: bool = False status: Literal["active", "paused", "done"] = "active" @field_validator("workspace_id", "topic", "active_goal", mode="before") @classmethod def _sanitize_text(cls, value: Any) -> str: return _clean_text(value, limit=240) @field_validator("pending_questions", "evidence_requirements", mode="before") @classmethod def _sanitize_list(cls, value: Any) -> list[str]: rows = value if isinstance(value, list) else [] return [_clean_text(item, limit=180) for item in rows[:12] if _clean_text(item, limit=180)] class PMFlowAIDisclosureState(BaseModel): """Customer-side disclosure boundary for this turn. This is a semantic state produced before action selection. Runtime code may consume it, but must not infer it from ad hoc keyword checks. """ visibility_class: DisclosureVisibilityClass = "public_product_capability" protected_categories: list[str] = Field(default_factory=list) allowed_response_mode: DisclosureResponseMode = "answer" policy_reason: str = Field(default="", max_length=400) contract_key: str = Field(default="", max_length=120) @field_validator("policy_reason", "contract_key", mode="before") @classmethod def _sanitize_text(cls, value: Any) -> str: return _clean_text(value, limit=400) @field_validator("protected_categories", mode="before") @classmethod def _sanitize_categories(cls, value: Any) -> list[str]: rows = value if isinstance(value, list) else [] return [_clean_text(item, limit=80) for item in rows[:12] if _clean_text(item, limit=80)] class PMFlowAIEvidenceGap(BaseModel): key: str = Field(..., min_length=1, max_length=120) label: str = Field(default="", max_length=240) severity: Literal["blocking", "non_blocking"] = "non_blocking" gap_type: GapType = "system_evidence_gap" user_actionable: bool = False customer_question: str = Field(default="", max_length=240) system_action_hint: str = Field(default="", max_length=160) reason: str = Field(default="", max_length=400) @field_validator("key", "label", "customer_question", "system_action_hint", "reason", mode="before") @classmethod def _sanitize_text(cls, value: Any) -> str: return _clean_text(value, limit=400) @model_validator(mode="after") def _sync_user_actionable_boundary(self) -> "PMFlowAIEvidenceGap": if self.gap_type in {"user_clarification_gap", "user_authorization_gap"}: self.user_actionable = True if self.gap_type == "system_evidence_gap": self.user_actionable = False self.customer_question = "" return self class PMFlowAIEvidenceFinding(BaseModel): fact: str = Field(..., min_length=1, max_length=700) source: str = Field(default="", max_length=240) implication: str = Field(default="", max_length=700) relevance_to_goal: str = Field(default="", max_length=160) confidence: EvidenceConfidence = "unknown" @field_validator("fact", "source", "implication", "relevance_to_goal", mode="before") @classmethod def _sanitize_text(cls, value: Any) -> str: return _clean_text(value, limit=700) class PMFlowAIEvidenceToolResult(BaseModel): packet_version: str = "pmflow_ai_evidence_tool_result.v1" tool_name: str = Field(..., min_length=1, max_length=120) action_type: RuntimeActionKind findings: list[PMFlowAIEvidenceFinding] = Field(default_factory=list) structured_evidence: dict[str, Any] = Field(default_factory=dict) sources: list[dict[str, Any]] = Field(default_factory=list) remaining_gaps: list[str] = Field(default_factory=list) confidence: EvidenceConfidence = "unknown" summary: str = Field(default="", max_length=500) @field_validator("tool_name", "summary", mode="before") @classmethod def _sanitize_text(cls, value: Any) -> str: return _clean_text(value, limit=500) @field_validator("remaining_gaps", mode="before") @classmethod def _sanitize_gaps(cls, value: Any) -> list[str]: rows = value if isinstance(value, list) else [] return [_clean_text(item, limit=160) for item in rows[:8] if _clean_text(item, limit=160)] class PMFlowAIEvidenceAbsorptionResult(BaseModel): packet_version: str = "pmflow_ai_evidence_absorption_result.v1" findings: list[PMFlowAIEvidenceFinding] = Field(default_factory=list) structured_evidence: dict[str, Any] = Field(default_factory=dict) sources: list[dict[str, Any]] = Field(default_factory=list) remaining_gaps: list[str] = Field(default_factory=list) judgement_state: PMFlowAIJudgementState = Field(default_factory=PMFlowAIJudgementState) evidence_quality: EvidenceConfidence = "unknown" summary: str = Field(default="", max_length=500) @field_validator("remaining_gaps", mode="before") @classmethod def _sanitize_gaps(cls, value: Any) -> list[str]: rows = value if isinstance(value, list) else [] return [_clean_text(item, limit=180) for item in rows[:10] if _clean_text(item, limit=180)] @field_validator("summary", mode="before") @classmethod def _sanitize_summary(cls, value: Any) -> str: return _clean_text(value, limit=500) class PMFlowAITurnEvidenceState(BaseModel): known_facts: list[PMFlowAIEvidenceFinding] = Field(default_factory=list) structured_evidence: dict[str, Any] = Field(default_factory=dict) evidence_gaps: list[PMFlowAIEvidenceGap] = Field(default_factory=list) evidence_quality: EvidenceConfidence = "unknown" sources: list[dict[str, Any]] = Field(default_factory=list) stale_or_missing: list[str] = Field(default_factory=list) @field_validator("stale_or_missing", mode="before") @classmethod def _sanitize_missing(cls, value: Any) -> list[str]: rows = value if isinstance(value, list) else [] return [_clean_text(item, limit=160) for item in rows[:8] if _clean_text(item, limit=160)] class PMFlowAIRuntimeAction(BaseModel): action_id: str = Field(..., min_length=1, max_length=80) action_type: RuntimeActionKind status: RuntimeActionStatus = "candidate" reason: str = Field(default="", max_length=320) tool_name: str = Field(default="", max_length=120) requires_confirmation: bool = False result_summary: str = Field(default="", max_length=500) @field_validator("action_id", "reason", "tool_name", "result_summary", mode="before") @classmethod def _sanitize_text(cls, value: Any) -> str: return _clean_text(value, limit=500) class PMFlowAITurnActionState(BaseModel): candidate_actions: list[PMFlowAIRuntimeAction] = Field(default_factory=list) performed_actions: list[PMFlowAIRuntimeAction] = Field(default_factory=list) blocked_actions: list[PMFlowAIRuntimeAction] = Field(default_factory=list) next_best_action: PMFlowAIRuntimeAction | None = None class PMFlowAISufficiencyState(BaseModel): can_answer_now: bool = False can_answer_with_assumptions: bool = False needs_user_input: bool = False needs_internal_evidence: bool = False needs_external_evidence: bool = False needs_confirmation_before_write: bool = False reason: str = Field(default="", max_length=400) @field_validator("reason", mode="before") @classmethod def _sanitize_reason(cls, value: Any) -> str: return _clean_text(value, limit=400) class PMFlowAIAnswerState(BaseModel): answer_intent: str = Field(default="answer", max_length=100) answer_shape: AnswerShape = "general_answer" required_sections: list[str] = Field(default_factory=list) prohibited_content: list[str] = Field(default_factory=list) citation_requirements: list[str] = Field(default_factory=list) @field_validator("answer_intent", mode="before") @classmethod def _sanitize_intent(cls, value: Any) -> str: return _clean_text(value, limit=100) @field_validator("required_sections", "prohibited_content", "citation_requirements", mode="before") @classmethod def _sanitize_list(cls, value: Any) -> list[str]: rows = value if isinstance(value, list) else [] return [_clean_text(item, limit=120) for item in rows[:10] if _clean_text(item, limit=120)] class PMFlowAIAnswerObjective(BaseModel): packet_version: str = "pmflow_ai_answer_objective.v1" primary_user_need: str = Field(default="", max_length=240) required_decision: str = Field(default="", max_length=240) expected_output: str = Field(default="", max_length=160) continuation_mode: str = Field(default="new_turn", max_length=80) must_include: list[str] = Field(default_factory=list) must_not_include: list[str] = Field(default_factory=list) evidence_to_use: list[str] = Field(default_factory=list) open_gaps: list[str] = Field(default_factory=list) confidence: EvidenceConfidence = "unknown" @field_validator("primary_user_need", "required_decision", "expected_output", "continuation_mode", mode="before") @classmethod def _sanitize_text(cls, value: Any) -> str: return _clean_text(value, limit=240) @field_validator("must_include", "must_not_include", "evidence_to_use", "open_gaps", mode="before") @classmethod def _sanitize_list(cls, value: Any) -> list[str]: rows = value if isinstance(value, list) else [] return [_clean_text(item, limit=180) for item in rows[:12] if _clean_text(item, limit=180)] class PMFlowAIAnswerPlan(BaseModel): packet_version: str = "pmflow_ai_answer_plan.v1" plan_id: str = Field(default="", max_length=80) answer_strategy: str = Field(default="", max_length=240) repeated_content_policy: str = Field(default="", max_length=240) boundary_statement_mode: str = Field(default="", max_length=120) evidence_usage_plan: list[str] = Field(default_factory=list) user_value_plan: list[str] = Field(default_factory=list) sections: list[str] = Field(default_factory=list) questions_to_ask_once: list[str] = Field(default_factory=list) draft_sources: list[str] = Field(default_factory=list) @field_validator("plan_id", "answer_strategy", "repeated_content_policy", "boundary_statement_mode", mode="before") @classmethod def _sanitize_text(cls, value: Any) -> str: return _clean_text(value, limit=240) @field_validator("evidence_usage_plan", "user_value_plan", "sections", "questions_to_ask_once", "draft_sources", mode="before") @classmethod def _sanitize_list(cls, value: Any) -> list[str]: rows = value if isinstance(value, list) else [] return [_clean_text(item, limit=180) for item in rows[:12] if _clean_text(item, limit=180)] class PMFlowAIAnswerDraft(BaseModel): packet_version: str = "pmflow_ai_answer_draft.v1" draft_id: str = Field(default="", max_length=80) source_renderer: str = Field(default="", max_length=120) answer_text: str = Field(default="", max_length=5000) plan_id: str = Field(default="", max_length=80) @field_validator("draft_id", "source_renderer", "answer_text", "plan_id", mode="before") @classmethod def _sanitize_text(cls, value: Any) -> str: return _clean_text(value, limit=5000) class PMFlowAIAnswerQualityFinding(BaseModel): finding_id: str = Field(default="", max_length=80) finding_type: str = Field(default="", max_length=120) reason: str = Field(default="", max_length=400) revision_instruction: str = Field(default="", max_length=400) blocking: bool = False @field_validator("finding_id", "finding_type", "reason", "revision_instruction", mode="before") @classmethod def _sanitize_text(cls, value: Any) -> str: return _clean_text(value, limit=400) class PMFlowAIRuntimeStep(BaseModel): key: str = Field(..., min_length=1, max_length=80) label: str = Field(..., min_length=1, max_length=160) status: RuntimeStepStatus = "completed" action_type: RuntimeActionKind | None = None duration_ms: int | None = Field(default=None, ge=0) evidence_delta: list[PMFlowAIEvidenceFinding] = Field(default_factory=list) blocking: bool = False @field_validator("key", "label", mode="before") @classmethod def _sanitize_text(cls, value: Any) -> str: return _clean_text(value, limit=160) class PMFlowAISufficiencyDecision(BaseModel): packet_version: str = "pmflow_ai_sufficiency_decision.v1" must_ask_user: bool = False can_answer_now: bool = False needs_action: bool = False next_action_type: RuntimeActionKind | None = None reason: str = Field(default="", max_length=400) @field_validator("reason", mode="before") @classmethod def _sanitize_reason(cls, value: Any) -> str: return _clean_text(value, limit=400) class PMFlowAITurnState(BaseModel): packet_version: str = PMFLOW_AI_TURN_STATE_VERSION visibility: Literal["customer_safe_trace"] = "customer_safe_trace" turn_goal: PMFlowAITurnGoal = Field(default_factory=PMFlowAITurnGoal) object_state: PMFlowAITurnObjectState = Field(default_factory=PMFlowAITurnObjectState) problem_frame: PMFlowAIProblemFrame = Field(default_factory=PMFlowAIProblemFrame) interpretation_state: PMFlowAIInterpretationState = Field(default_factory=PMFlowAIInterpretationState) evidence_state: PMFlowAITurnEvidenceState = Field(default_factory=PMFlowAITurnEvidenceState) judgement_state: PMFlowAIJudgementState = Field(default_factory=PMFlowAIJudgementState) action_state: PMFlowAITurnActionState = Field(default_factory=PMFlowAITurnActionState) sufficiency_state: PMFlowAISufficiencyState = Field(default_factory=PMFlowAISufficiencyState) answer_state: PMFlowAIAnswerState = Field(default_factory=PMFlowAIAnswerState) answer_objective: PMFlowAIAnswerObjective = Field(default_factory=PMFlowAIAnswerObjective) answer_plan: PMFlowAIAnswerPlan = Field(default_factory=PMFlowAIAnswerPlan) answer_draft: PMFlowAIAnswerDraft = Field(default_factory=PMFlowAIAnswerDraft) answer_quality_findings: list[PMFlowAIAnswerQualityFinding] = Field(default_factory=list) terminal_answer_state: PMFlowAITerminalAnswerState = Field(default_factory=PMFlowAITerminalAnswerState) disclosure_state: PMFlowAIDisclosureState = Field(default_factory=PMFlowAIDisclosureState) runtime_steps: list[PMFlowAIRuntimeStep] = Field(default_factory=list) route_plan_input: dict[str, Any] = Field(default_factory=dict) legacy_runtime_decision: dict[str, Any] = Field(default_factory=dict) max_steps: int = Field(default=3, ge=1, le=6) completed: bool = False thread_workspace: PMFlowAIThreadWorkspace | None = None turn_delta: PMFlowAITurnDelta | None = None candidate_frames: list[PMFlowAIGoalFrame] = Field(default_factory=list) selected_frame: PMFlowAIGoalFrame | None = None observations: list[PMFlowAIObservation] = Field(default_factory=list) operation_gaps: list[PMFlowAIOperationGap] = Field(default_factory=list) state_revisions: list[PMFlowAIStateRevision] = Field(default_factory=list) verification_failures: list[PMFlowAIVerificationFailure] = Field(default_factory=list) customer_packet_draft: dict[str, Any] = Field(default_factory=dict) step_budget: int = Field(default=6, ge=1, le=12) @model_validator(mode="after") def _cap_lists(self) -> "PMFlowAITurnState": self.runtime_steps = list(self.runtime_steps or [])[:12] self.evidence_state.known_facts = list(self.evidence_state.known_facts or [])[:20] self.candidate_frames = list(self.candidate_frames or [])[:8] self.observations = list(self.observations or [])[:20] self.operation_gaps = list(self.operation_gaps or [])[:20] self.state_revisions = list(self.state_revisions or [])[:20] self.verification_failures = list(self.verification_failures or [])[:12] self.answer_quality_findings = list(self.answer_quality_findings or [])[:12] return self __all__ = [ "PMFLOW_AI_TURN_STATE_VERSION", "AnswerShape", "EvidenceConfidence", "GapType", "ObservationType", "OperationKind", "SourcePolicy", "PMFlowAIGoalFrame", "PMFlowAIAnswerDraft", "PMFlowAIAnswerObjective", "PMFlowAIAnswerPlan", "PMFlowAIAnswerQualityFinding", "PMFlowAIAnswerState", "PMFlowAIAnswerIntentPlan", "PMFlowAIAttempt", "PMFlowAIBoundary", "PMFlowAIDisclosureState", "PMFlowAIClaim", "PMFlowAIEvidenceFinding", "PMFlowAIEvidenceGap", "PMFlowAIEvidenceAbsorptionResult", "PMFlowAIEvidenceToolResult", "PMFlowAIInterpretationBranch", "PMFlowAIInterpretationState", "PMFlowAIJudgementState", "PMFlowAIOperationGap", "PMFlowAIObservation", "PMFlowAIObservationItem", "PMFlowAIProblemFrame", "PMFlowAIRuntimeAction", "PMFlowAIRuntimeStep", "PMFlowAIStateRevision", "PMFlowAISufficiencyDecision", "PMFlowAISufficiencyState", "PMFlowAITargetCollection", "PMFlowAITargetScope", "PMFlowAITerminalAnswerState", "PMFlowAITurnActionState", "PMFlowAITurnEvidenceState", "PMFlowAITurnGoal", "PMFlowAITurnObjectRef", "PMFlowAITurnObjectState", "PMFlowAITurnState", "PMFlowAITurnDelta", "PMFlowAITurnDeltaAnswer", "PMFlowAIThreadWorkspace", "PMFlowAIVerificationFailure", "RuntimeActionPhase", "RuntimeActionKind", "TargetObjectRequirement", "TargetScopeBindingStrength", "TargetScopeType", "ThreadWorkspaceKind", "DisclosureResponseMode", "DisclosureVisibilityClass", ]