"""Operating cycle triggers, outcome observation and learning ledger.""" from __future__ import annotations import hashlib from typing import Any from ...schemas.pmflow_ai_operating_runtime import ( PMFlowAIAgentCouncilReview, PMFlowAIAutonomyDecision, PMFlowAILearningLedger, PMFlowAIOperatingCycleState, PMFlowAIOperatingCycleTrigger, PMFlowAIOperatingMission, PMFlowAIOperatingPlan, PMFlowAIOutcomeObservation, PMFlowAIProjectWorldStatePacket, clean_text, ) from ...schemas.rag import RAGAskRequest SUPPORTED_OPERATING_TRIGGERS = [ { "trigger_type": "user_turn", "display_label": "用户对话", "contract": "run_immediately_from_customer_turn", }, { "trigger_type": "signal", "display_label": "项目/市场信号", "contract": "run_when_new_signal_changes_project_context", }, { "trigger_type": "project_creation", "display_label": "项目创建", "contract": "run_after_project_truth_state_created", }, { "trigger_type": "task_update", "display_label": "任务更新", "contract": "run_after_task_or_changeset_state_changes", }, { "trigger_type": "execution_result", "display_label": "执行结果", "contract": "run_after_execution_bridge_reports_result", }, { "trigger_type": "daily_review", "display_label": "每日复盘", "contract": "scheduled_operating_review", }, { "trigger_type": "weekly_strategy_review", "display_label": "每周策略复盘", "contract": "scheduled_strategy_review", }, ] class PMFlowAIOperatingCycleScheduler: """Build trigger contracts for operating runs.""" def build_trigger( self, *, request: RAGAskRequest, mission: PMFlowAIOperatingMission, trigger_type: str = "user_turn", source_payload: dict[str, Any] | None = None, ) -> PMFlowAIOperatingCycleTrigger: normalized_type = clean_text(trigger_type, limit=80) supported = {item["trigger_type"] for item in SUPPORTED_OPERATING_TRIGGERS} if normalized_type not in supported: normalized_type = "user_turn" project_ref = mission.project_scope.object_refs[0] if mission.project_scope.object_refs else "" source_ref = f"qa_session:{request.session_id}" if request.session_id is not None else "space_ai_turn" return PMFlowAIOperatingCycleTrigger( trigger_type=normalized_type, source_ref=source_ref, project_ref=project_ref, payload={ "question_ref": _stable_question_ref(request.question), "asked_via": clean_text(request.asked_via or "pmflow", limit=40), "source_payload": dict(source_payload or {}), }, ) def supported_triggers(self) -> list[dict[str, Any]]: return [dict(item) for item in SUPPORTED_OPERATING_TRIGGERS] def next_cycle_triggers( self, *, mission: PMFlowAIOperatingMission, autonomy_gate: PMFlowAIAutonomyDecision, learning_ledger: PMFlowAILearningLedger, ) -> list[dict[str, Any]]: project_ref = mission.project_scope.object_refs[0] if mission.project_scope.object_refs else "" triggers = [ { "trigger_type": "user_turn", "reason": "用户继续追问、确认或补充后重新运行 operating cycle。", "project_ref": project_ref, }, { "trigger_type": "execution_result", "reason": "执行交接或任务结果返回后更新 learning ledger 并重排计划。", "project_ref": project_ref, }, { "trigger_type": "daily_review", "reason": "每日读取项目运行态和学习账本，检查阻塞、确认项和下一步。", "project_ref": project_ref, }, { "trigger_type": "weekly_strategy_review", "reason": "每周基于结果、偏好和实验信号重新判断方向。", "project_ref": project_ref, }, ] if autonomy_gate.confirmation_requests: triggers.append( { "trigger_type": "task_update", "reason": "确认请求被接受/拒绝后更新决策结果。", "project_ref": project_ref, } ) if learning_ledger.experiment_results: triggers.append( { "trigger_type": "signal", "reason": "实验结果或市场信号进入后刷新客户验证判断。", "project_ref": project_ref, } ) return triggers[:8] class OutcomeObserver: """Observe decisions, execution results, experiments and preferences.""" def observe( self, *, mission: PMFlowAIOperatingMission, project_world_state: PMFlowAIProjectWorldStatePacket, council_review: PMFlowAIAgentCouncilReview, autonomy_gate: PMFlowAIAutonomyDecision, operating_plan: PMFlowAIOperatingPlan, external_outcomes: list[dict[str, Any]] | None = None, ) -> list[PMFlowAIOutcomeObservation]: observations: list[PMFlowAIOutcomeObservation] = [] observations.extend( DecisionOutcomeService().observe_pending_confirmations( mission=mission, autonomy_gate=autonomy_gate, ) ) observations.extend( ExperimentResultService().observe_pending_experiments( autonomy_gate=autonomy_gate, ) ) observations.extend( ExecutionOutcomeService().observe_execution_preparations( autonomy_gate=autonomy_gate, ) ) observations.extend( UserPreferenceLearningService().observe_preferences( mission=mission, operating_plan=operating_plan, council_review=council_review, ) ) observations.extend(self.from_external_outcomes(external_outcomes or [], project_world_state=project_world_state)) return observations[:24] def from_external_outcomes( self, rows: list[dict[str, Any]], *, project_world_state: PMFlowAIProjectWorldStatePacket, ) -> list[PMFlowAIOutcomeObservation]: observations: list[PMFlowAIOutcomeObservation] = [] for item in rows[:24]: if not isinstance(item, dict): continue outcome_type = clean_text(item.get("outcome_type"), limit=80) if outcome_type not in {"recommendation_decision", "execution_result", "experiment_result", "user_preference"}: continue observations.append( PMFlowAIOutcomeObservation( outcome_type=outcome_type, outcome_status=_normalize_outcome_status(item.get("outcome_status") or item.get("status")), source_ref=clean_text(item.get("source_ref"), limit=180), target_ref=clean_text(item.get("target_ref") or project_world_state.primary_project_ref, limit=160), summary=clean_text(item.get("summary") or item.get("title"), limit=800), evidence_refs=[clean_text(ref, limit=220) for ref in list(item.get("evidence_refs") or [])[:16]], learning_signals=dict(item.get("learning_signals") or {}) if isinstance(item.get("learning_signals"), dict) else {}, ) ) return observations class DecisionOutcomeService: def observe_pending_confirmations( self, *, mission: PMFlowAIOperatingMission, autonomy_gate: PMFlowAIAutonomyDecision, ) -> list[PMFlowAIOutcomeObservation]: observations: list[PMFlowAIOutcomeObservation] = [] project_ref = mission.project_scope.object_refs[0] if mission.project_scope.object_refs else "" for request in list(autonomy_gate.confirmation_requests or [])[:12]: if not isinstance(request, dict): continue observations.append( PMFlowAIOutcomeObservation( outcome_type="recommendation_decision", outcome_status="pending", source_ref=clean_text(request.get("confirmation_request_id"), limit=120), target_ref=clean_text(request.get("target_ref") or project_ref, limit=160), summary=f"确认请求待用户决策：{clean_text(request.get('title'), limit=240)}", learning_signals={ "decision_state": "pending_user_confirmation", "requested_for": list(request.get("requested_for") or []), }, ) ) return observations class ExecutionOutcomeService: def observe_execution_preparations( self, *, autonomy_gate: PMFlowAIAutonomyDecision, ) -> list[PMFlowAIOutcomeObservation]: observations: list[PMFlowAIOutcomeObservation] = [] for item in list(autonomy_gate.execution_preparations or [])[:8]: if not isinstance(item, dict): continue observations.append( PMFlowAIOutcomeObservation( outcome_type="execution_result", outcome_status="pending", source_ref=clean_text(item.get("execution_preparation_id"), limit=120), target_ref=clean_text(item.get("target_ref"), limit=160), summary="执行交接包已准备，但尚未发起 handoff，等待用户确认。", learning_signals={ "execution_state": "prepared_waiting_confirmation", "readiness": clean_text(item.get("readiness"), limit=80), }, ) ) return observations class ExperimentResultService: def observe_pending_experiments( self, *, autonomy_gate: PMFlowAIAutonomyDecision, ) -> list[PMFlowAIOutcomeObservation]: observations: list[PMFlowAIOutcomeObservation] = [] for item in list(autonomy_gate.prepared_action_packets or [])[:12]: if not isinstance(item, dict) or item.get("action_family") != "experiment_plan": continue observations.append( PMFlowAIOutcomeObservation( outcome_type="experiment_result", outcome_status="pending", source_ref=clean_text(item.get("prepared_action_id"), limit=120), target_ref=clean_text(item.get("target_ref"), limit=160), summary=f"用户验证实验待确认：{clean_text(item.get('title'), limit=240)}", evidence_refs=list(item.get("evidence_refs") or []), learning_signals={ "experiment_state": "prepared_waiting_confirmation", "action_family": item.get("action_family"), }, ) ) return observations class UserPreferenceLearningService: def observe_preferences( self, *, mission: PMFlowAIOperatingMission, operating_plan: PMFlowAIOperatingPlan, council_review: PMFlowAIAgentCouncilReview, ) -> list[PMFlowAIOutcomeObservation]: project_ref = mission.project_scope.object_refs[0] if mission.project_scope.object_refs else "" observations = [ PMFlowAIOutcomeObservation( outcome_type="user_preference", outcome_status="observed", source_ref=mission.mission_id, target_ref=project_ref, summary="用户偏好体现为：需要 PMFlow AI 主动推进项目，而不是只回答问题。", learning_signals={ "preference_key": "operating_partner_not_qa", "preferred_behavior": "prepare_plan_with_confirmation_gate", "risk_threshold": "do_not_execute_without_explicit_confirmation", }, ) ] if operating_plan.priority_order: observations.append( PMFlowAIOutcomeObservation( outcome_type="user_preference", outcome_status="observed", source_ref=str(council_review.review_id), target_ref=project_ref, summary="本轮计划保留单一优先级，后续运行应继续消费 priority_order 而不是散列建议。", learning_signals={ "preference_key": "single_priority_order", "preferred_behavior": "compile_one_ranked_plan", "source_effectiveness_policy": "future_runs_should_reuse_evidence_types_that_survive_council_review", }, ) ) if operating_plan.product_object_projections: observations.append( PMFlowAIOutcomeObservation( outcome_type="user_preference", outcome_status="observed", source_ref=mission.mission_id, target_ref=project_ref, summary="本轮已把回答升级为经营对象候选；后续应跟踪用户是否保存、确认、修改或拒绝该对象。", learning_signals={ "preference_key": "product_object_first", "preferred_behavior": "project_opportunity_or_research_outputs_before_freeform_answer", }, ) ) return observations class LearningLedgerService: """Load and update durable learning from session capsules.""" def ledger_from_context( self, *, conversation_context: dict[str, Any] | None, mission: PMFlowAIOperatingMission, ) -> PMFlowAILearningLedger: capsules = list((conversation_context or {}).get("context_capsules") or []) project_ref = mission.project_scope.object_refs[0] if mission.project_scope.object_refs else "" for item in reversed(capsules): if not isinstance(item, dict) or item.get("capsule_type") != "operating_learning_ledger": continue ledger = item.get("learning_ledger") if isinstance(item.get("learning_ledger"), dict) else {} if not ledger: continue if project_ref and ledger.get("scope_ref") not in {"", project_ref}: continue return PMFlowAILearningLedger.model_validate(ledger) return PMFlowAILearningLedger(scope_ref=project_ref) def update_ledger( self, *, consumed_ledger: PMFlowAILearningLedger, outcome_observations: list[PMFlowAIOutcomeObservation], ) -> PMFlowAILearningLedger: decision_outcomes = list(consumed_ledger.decision_outcomes or []) experiment_results = list(consumed_ledger.experiment_results or []) execution_outcomes = list(consumed_ledger.execution_outcomes or []) user_preferences = list(consumed_ledger.user_preferences or []) learning_items = list(consumed_ledger.learning_items or []) for observation in outcome_observations: row = _observation_row(observation) if observation.outcome_type == "recommendation_decision": decision_outcomes.append(row) elif observation.outcome_type == "experiment_result": experiment_results.append(row) elif observation.outcome_type == "execution_result": execution_outcomes.append(row) elif observation.outcome_type == "user_preference": user_preferences.append(row) learning_items.append(_learning_item_from_observation(observation)) return PMFlowAILearningLedger( ledger_id=consumed_ledger.ledger_id, scope_ref=consumed_ledger.scope_ref, learning_items=_dedupe_rows(learning_items, key="learning_key")[-40:], decision_outcomes=_dedupe_rows(decision_outcomes, key="source_ref")[-40:], experiment_results=_dedupe_rows(experiment_results, key="source_ref")[-40:], execution_outcomes=_dedupe_rows(execution_outcomes, key="source_ref")[-40:], user_preferences=_dedupe_rows(user_preferences, key="learning_key")[-40:], consumed_memory_refs=_dedupe_texts( list(consumed_ledger.consumed_memory_refs or []) + [consumed_ledger.ledger_id] )[-24:], ) def capsule(self, *, learning_ledger: PMFlowAILearningLedger) -> dict[str, Any]: return { "capsule_version": "pmflow_ai_learning_ledger_capsule.v1", "capsule_type": "operating_learning_ledger", "scope_ref": learning_ledger.scope_ref, "ledger_id": learning_ledger.ledger_id, "updated_at": learning_ledger.updated_at, "learning_ledger": learning_ledger.model_dump(mode="json", exclude_none=True), } class OperatingCycleService: def __init__(self) -> None: self.scheduler = PMFlowAIOperatingCycleScheduler() self.observer = OutcomeObserver() self.ledger_service = LearningLedgerService() def build_cycle_state( self, *, request: RAGAskRequest, mission: PMFlowAIOperatingMission, project_world_state: PMFlowAIProjectWorldStatePacket, council_review: PMFlowAIAgentCouncilReview, autonomy_gate: PMFlowAIAutonomyDecision, operating_plan: PMFlowAIOperatingPlan, conversation_context: dict[str, Any] | None, ) -> PMFlowAIOperatingCycleState: consumed_ledger = self.ledger_service.ledger_from_context( conversation_context=conversation_context, mission=mission, ) trigger = self.scheduler.build_trigger(request=request, mission=mission) external_outcomes = _external_outcomes_from_context(conversation_context) observations = self.observer.observe( mission=mission, project_world_state=project_world_state, council_review=council_review, autonomy_gate=autonomy_gate, operating_plan=operating_plan, external_outcomes=external_outcomes, ) learning_ledger = self.ledger_service.update_ledger( consumed_ledger=consumed_ledger, outcome_observations=observations, ) return PMFlowAIOperatingCycleState( trigger=trigger, supported_triggers=self.scheduler.supported_triggers(), consumed_learning_ledger=consumed_ledger, outcome_observations=observations, learning_ledger=learning_ledger, next_cycle_triggers=self.scheduler.next_cycle_triggers( mission=mission, autonomy_gate=autonomy_gate, learning_ledger=learning_ledger, ), ) def _external_outcomes_from_context(conversation_context: dict[str, Any] | None) -> list[dict[str, Any]]: context = conversation_context or {} rows = context.get("operating_outcomes") return [dict(item) for item in rows if isinstance(item, dict)] if isinstance(rows, list) else [] def _normalize_outcome_status(value: Any) -> str: status = clean_text(value, limit=80) if status in {"pending", "accepted", "rejected", "succeeded", "failed", "partial", "observed"}: return status return "observed" def _observation_row(observation: PMFlowAIOutcomeObservation) -> dict[str, Any]: return { "observation_id": observation.observation_id, "outcome_type": observation.outcome_type, "outcome_status": observation.outcome_status, "source_ref": observation.source_ref, "target_ref": observation.target_ref, "summary": observation.summary, "learning_signals": dict(observation.learning_signals or {}), "observed_at": observation.observed_at, } def _learning_item_from_observation(observation: PMFlowAIOutcomeObservation) -> dict[str, Any]: signals = dict(observation.learning_signals or {}) learning_key = clean_text(signals.get("preference_key") or observation.source_ref or observation.outcome_type, limit=180) return { "learning_key": learning_key, "outcome_type": observation.outcome_type, "outcome_status": observation.outcome_status, "summary": observation.summary, "target_ref": observation.target_ref, "updated_at": observation.observed_at, } def _stable_question_ref(question: Any) -> str: text = clean_text(question, limit=80) if not text: return "question:empty" digest = hashlib.sha1(text.encode("utf-8")).hexdigest()[:12] return f"question:{digest}" def _dedupe_rows(rows: list[dict[str, Any]], *, key: str) -> list[dict[str, Any]]: seen: set[str] = set() out: list[dict[str, Any]] = [] for row in rows: if not isinstance(row, dict): continue value = clean_text(row.get(key), limit=220) if not value: value = clean_text(row.get("summary"), limit=220) marker = value.lower() if not marker or marker in seen: continue seen.add(marker) out.append(row) return out def _dedupe_texts(rows: list[str]) -> list[str]: out: list[str] = [] for item in rows: text = clean_text(item, limit=220) if text and text not in out: out.append(text) return out __all__ = [ "DecisionOutcomeService", "ExecutionOutcomeService", "ExperimentResultService", "LearningLedgerService", "OperatingCycleService", "OutcomeObserver", "PMFlowAIOperatingCycleScheduler", "UserPreferenceLearningService", ]