PMFlow/backend/app/services/ai_pmo/semantic_context/decision_schema.py

128 lines
4.5 KiB
Python

from __future__ import annotations
from typing import Any, Literal
from pydantic import BaseModel, ConfigDict, Field
from .business_frame_schema import (
AnswerShapeContract,
BusinessSemanticFrame,
OperationIntentContract,
)
BoundaryLevel = Literal["status", "recommendation", "draft", "execute"]
class ModeCard(BaseModel):
model_config = ConfigDict(extra="ignore")
workflow_mode: Literal["CHAT", "PLAN", "EXECUTE"] = "CHAT"
mode_state: str = "IDLE"
allowed_actions: list[str] = Field(default_factory=list)
forbidden_actions: list[str] = Field(default_factory=list)
exit_conditions: list[str] = Field(default_factory=list)
class DecisionWorkflowPlan(BaseModel):
model_config = ConfigDict(extra="ignore")
workflow_type: Literal[
"query_flow",
"new_project_flow",
"project_flow",
"task_flow",
"signal_source_flow",
"project_signal_flow",
"followup_flow",
]
workflow_action: str
target_scope: Literal["global", "project", "thread"] = "global"
operation_codes: list[str] = Field(default_factory=list)
requires_confirmation: bool = False
confidence: float = 0.0
clarification_needed: bool = False
metadata: dict[str, Any] = Field(default_factory=dict)
class InteractionBoundaryPacket(BaseModel):
model_config = ConfigDict(extra="ignore")
boundary_version: str = "v1"
requested_level: BoundaryLevel | Literal["unclear"] = "unclear"
forbidden_levels: list[BoundaryLevel] = Field(default_factory=list)
allow_status: bool = True
allow_recommendation: bool = True
allow_draft: bool = False
allow_execute: bool = False
allow_write: bool = False
needs_clarification: bool = False
clarification_reason: str = ""
clarification_questions: list[str] = Field(default_factory=list)
correction_allowed: bool = True
correction_targets: list[Literal["project", "object", "level", "priority", "assumption"]] = Field(
default_factory=list
)
self_review_required: bool = True
confidence: float = 0.0
evidence: list[dict[str, Any]] = Field(default_factory=list)
class AnswerControlDecision(BaseModel):
model_config = ConfigDict(extra="ignore")
effective_level: BoundaryLevel | Literal["clarify"]
blocked_levels: list[BoundaryLevel] = Field(default_factory=list)
blocked_workflows: list[str] = Field(default_factory=list)
allowed_workflows: list[str] = Field(default_factory=list)
downgraded_from: BoundaryLevel | None = None
requires_confirmation: bool = False
requires_self_review: bool = True
requires_correction_slot: bool = True
reason: str = ""
class SelfReviewResult(BaseModel):
model_config = ConfigDict(extra="ignore")
project_aligned: bool = True
object_aligned: bool = True
level_aligned: bool = True
leaked_internal_terms: bool = False
overstepped_boundary: bool = False
clarify_missing: bool = False
notes: list[str] = Field(default_factory=list)
class SemanticDecision(BaseModel):
model_config = ConfigDict(extra="ignore")
topic_relation: Literal["continue_current_topic", "new_topic", "uncertain"]
mode_decision: Literal["stay_plan", "ready_confirm", "request_execute", "answer_only"]
followup_scope: Literal["none", "current_requirement", "current_draft", "independent_workflow", "clarify"] = "none"
followup_action: Literal["none", "confirm", "revise", "regenerate", "cancel", "answer_only"] = "none"
revision_note: str = ""
draft_patch: dict[str, Any] = Field(default_factory=dict)
workflow_plans: list[DecisionWorkflowPlan] = Field(default_factory=list)
scene_metadata: dict[str, Any] = Field(default_factory=dict)
interaction_boundary: InteractionBoundaryPacket | None = None
business_frame: BusinessSemanticFrame | None = None
answer_shape_contract: AnswerShapeContract | None = None
operation_intent_contract: OperationIntentContract | None = None
self_review: SelfReviewResult | None = None
assistant_reply: str = ""
requires_confirmation: bool = False
requested_actions: list[str] = Field(default_factory=list)
confidence: float = 0.0
uncertainties: list[str] = Field(default_factory=list)
class DecisionGateResult(BaseModel):
model_config = ConfigDict(extra="ignore")
accepted: bool
blocked_reasons: list[str] = Field(default_factory=list)
allowed_actions: list[str] = Field(default_factory=list)
next_plan_status: str
needs_confirmation: bool = False
mode_state_after: str | None = None