774 lines
37 KiB
Python
774 lines
37 KiB
Python
"""Truth-bound project copilot answer service."""
|
||
|
||
from __future__ import annotations
|
||
|
||
from typing import Any
|
||
|
||
from sqlalchemy.ext.asyncio import AsyncSession
|
||
|
||
from ..schemas.pmflow_os import (
|
||
ProjectCopilotActionLadderLevel,
|
||
ProjectCopilotAllowedAction,
|
||
ProjectCopilotAnswerCanvas,
|
||
ProjectCopilotAskRequest,
|
||
ProjectCopilotAskResponse,
|
||
ProjectCopilotContextPacket,
|
||
ProjectCopilotObjectAnchorPacket,
|
||
ProjectTruthStateProjection,
|
||
)
|
||
from .pmflow_os_projection_service import PMFlowOSProjectionService
|
||
|
||
|
||
class ProjectCopilotService:
|
||
"""Build customer-safe project copilot answers from the project truth projection.
|
||
|
||
This service intentionally stays read-only. It may explain, draft candidate direction,
|
||
review risk, prepare handoff guidance, or route the user back to a PMFlow surface, but
|
||
it never commits a ChangeSet, writes Project Memory, executes commands, or publishes.
|
||
"""
|
||
|
||
def __init__(self, db: AsyncSession, *, tenant_id: int):
|
||
self.db = db
|
||
self.tenant_id = int(tenant_id)
|
||
self.projection_service = PMFlowOSProjectionService(db, tenant_id=tenant_id)
|
||
|
||
async def ask(
|
||
self,
|
||
*,
|
||
project_id: int,
|
||
current_user_id: int,
|
||
body: ProjectCopilotAskRequest,
|
||
) -> ProjectCopilotAskResponse:
|
||
truth_state = await self.projection_service.get_truth_state(
|
||
project_id=int(project_id),
|
||
current_user_id=int(current_user_id),
|
||
)
|
||
try:
|
||
execution_summary = await self.projection_service.get_execution_summary(
|
||
project_id=int(project_id),
|
||
current_user_id=int(current_user_id),
|
||
)
|
||
except Exception:
|
||
execution_summary = None
|
||
|
||
anchor = self._resolve_anchor(truth_state, body)
|
||
packet = self._build_context_packet(
|
||
truth_state,
|
||
body=body,
|
||
anchor=anchor,
|
||
execution_summary=execution_summary.model_dump(mode="json") if execution_summary else {},
|
||
)
|
||
canvas = self._render_answer(body=body, packet=packet, truth_state=truth_state)
|
||
return ProjectCopilotAskResponse(
|
||
answer_canvas=canvas,
|
||
project_copilot_context_packet=packet,
|
||
customer_interaction_context_packet={
|
||
"visible_objects": [
|
||
{
|
||
"object_ref": f"{packet.object_anchor.type}:{packet.object_anchor.id}",
|
||
"display_label": packet.object_anchor.title,
|
||
"object_type": packet.object_anchor.type,
|
||
"surface": packet.surface,
|
||
}
|
||
],
|
||
"current_focus_object": {
|
||
"object_ref": f"{packet.object_anchor.type}:{packet.object_anchor.id}",
|
||
"display_label": packet.object_anchor.title,
|
||
"surface": packet.surface,
|
||
},
|
||
},
|
||
answer_process_packet={
|
||
"steps": [
|
||
{"label": "读取项目真相", "status": "completed", "blocking": True},
|
||
{"label": "绑定当前对象锚点", "status": "completed", "blocking": True},
|
||
{"label": "校验允许动作和禁止动作", "status": "completed", "blocking": True},
|
||
{"label": "生成低风险主动建议", "status": "completed", "blocking": False},
|
||
{"label": "生成客户安全回答", "status": "completed", "blocking": False},
|
||
],
|
||
"model_calls": 0,
|
||
"source": "project_copilot_deterministic_renderer",
|
||
},
|
||
allowed_actions=packet.allowed_actions,
|
||
blocked_actions=packet.forbidden_actions,
|
||
proactive_suggestions=packet.proactive_suggestions,
|
||
warnings=[],
|
||
)
|
||
|
||
def _resolve_anchor(
|
||
self,
|
||
truth_state: ProjectTruthStateProjection,
|
||
body: ProjectCopilotAskRequest,
|
||
) -> ProjectCopilotObjectAnchorPacket:
|
||
requested = body.object_anchor
|
||
if requested:
|
||
anchor = self._anchor_from_request(truth_state, requested.type, requested.id)
|
||
if anchor:
|
||
return anchor
|
||
return self._default_anchor(truth_state, body.surface)
|
||
|
||
def _anchor_from_request(
|
||
self,
|
||
truth_state: ProjectTruthStateProjection,
|
||
anchor_type: str,
|
||
anchor_id: str,
|
||
) -> ProjectCopilotObjectAnchorPacket | None:
|
||
if not anchor_id:
|
||
return None
|
||
if anchor_type == "change_set":
|
||
item = _find_by_id(truth_state.changeSets, anchor_id)
|
||
if item:
|
||
return ProjectCopilotObjectAnchorPacket(
|
||
type="change_set",
|
||
id=str(item.get("id", anchor_id)),
|
||
title=str(item.get("headline") or item.get("summary") or f"变更集 {anchor_id}"),
|
||
summary=str(item.get("summary") or "当前变更集。"),
|
||
status_label=_change_set_status_label(item.get("status")),
|
||
mode="committed" if str(item.get("status")) in {"committed", "partially_committed"} else "current",
|
||
source_label="项目待确认链",
|
||
risk_level=self._risk_for_change_set(truth_state, str(item.get("id", anchor_id))),
|
||
evidence_ids=self._evidence_for_change_set(truth_state, str(item.get("id", anchor_id))),
|
||
)
|
||
if anchor_type == "change_item":
|
||
item = _find_by_id(truth_state.changeItems, anchor_id)
|
||
if item:
|
||
return ProjectCopilotObjectAnchorPacket(
|
||
type="change_item",
|
||
id=str(item.get("id", anchor_id)),
|
||
title=str(item.get("title") or f"变更项 {anchor_id}"),
|
||
summary=str(item.get("summary") or item.get("diff") or "当前变更项。"),
|
||
status_label=_change_item_status_label(item.get("status")),
|
||
mode="committed" if str(item.get("status")) in {"committed", "skipped"} else "current",
|
||
source_label="项目待确认链",
|
||
risk_level=_risk_level(item.get("riskLevel") or item.get("risk_level")),
|
||
evidence_ids=[str(eid) for eid in item.get("evidenceIds", []) or item.get("evidence_ids", []) or []],
|
||
)
|
||
if anchor_type == "task":
|
||
task = next((item for item in truth_state.tasks if str(item.id) == str(anchor_id)), None)
|
||
if task:
|
||
return ProjectCopilotObjectAnchorPacket(
|
||
type="task",
|
||
id=str(task.id),
|
||
title=task.title,
|
||
summary=task.summary,
|
||
status_label=_task_status_label(task.status),
|
||
mode="historical" if task.status in {"done", "blocked"} else "current",
|
||
source_label="项目工作图 / 开发执行",
|
||
risk_level=self._risk_for_stage(truth_state, task.stage),
|
||
evidence_ids=[],
|
||
)
|
||
if anchor_type == "release":
|
||
release = next((item for item in truth_state.releases if str(item.id) == str(anchor_id)), None)
|
||
if release:
|
||
return ProjectCopilotObjectAnchorPacket(
|
||
type="release",
|
||
id=str(release.id),
|
||
title=release.title,
|
||
summary=release.summary,
|
||
status_label=_release_status_label(release.status),
|
||
mode="current" if release.status in {"ready", "shipping", "live", "watch"} else "stale",
|
||
source_label="发布与增长",
|
||
risk_level="medium",
|
||
evidence_ids=[],
|
||
)
|
||
if anchor_type == "evidence":
|
||
evidence = _find_by_id(truth_state.evidence, anchor_id)
|
||
if evidence:
|
||
return ProjectCopilotObjectAnchorPacket(
|
||
type="evidence",
|
||
id=str(evidence.get("id", anchor_id)),
|
||
title=str(evidence.get("title") or f"证据 {anchor_id}"),
|
||
summary=str(evidence.get("excerpt") or evidence.get("summary") or "当前证据。"),
|
||
status_label="可引用",
|
||
mode="current",
|
||
source_label=str(evidence.get("sourceLabel") or evidence.get("source_label") or "项目证据"),
|
||
risk_level="low",
|
||
evidence_ids=[str(evidence.get("id", anchor_id))],
|
||
)
|
||
if anchor_type == "project" and str(truth_state.project.id) == str(anchor_id):
|
||
return self._project_anchor(truth_state)
|
||
return None
|
||
|
||
def _default_anchor(self, truth_state: ProjectTruthStateProjection, surface: str) -> ProjectCopilotObjectAnchorPacket:
|
||
pending_items = [item for item in truth_state.changeItems if str(item.get("status")) == "pending_review"]
|
||
if surface == "change_review" and pending_items:
|
||
return self._anchor_from_request(truth_state, "change_item", str(pending_items[0].get("id"))) or self._project_anchor(truth_state)
|
||
if surface in {"execution_bridge", "work_graph"} and truth_state.tasks:
|
||
selected = next((item for item in truth_state.tasks if str(item.id) == str(truth_state.selectedTaskId)), None) or truth_state.tasks[0]
|
||
return self._anchor_from_request(truth_state, "task", str(selected.id)) or self._project_anchor(truth_state)
|
||
if surface == "release_growth" and truth_state.releases:
|
||
return self._anchor_from_request(truth_state, "release", str(truth_state.releases[0].id)) or self._project_anchor(truth_state)
|
||
if truth_state.activeChangeSetId:
|
||
anchor = self._anchor_from_request(truth_state, "change_set", truth_state.activeChangeSetId)
|
||
if anchor:
|
||
return anchor
|
||
return self._project_anchor(truth_state)
|
||
|
||
def _project_anchor(self, truth_state: ProjectTruthStateProjection) -> ProjectCopilotObjectAnchorPacket:
|
||
high_risk = any(getattr(item, "level", "low") == "high" for item in truth_state.risks)
|
||
return ProjectCopilotObjectAnchorPacket(
|
||
type="project",
|
||
id=str(truth_state.project.id),
|
||
title=truth_state.project.name,
|
||
summary=truth_state.project.goal,
|
||
status_label=_stage_label(truth_state.project.stage),
|
||
mode="current",
|
||
source_label="项目真相层",
|
||
risk_level="high" if high_risk else "low",
|
||
evidence_ids=[str(item.get("id")) for item in truth_state.evidence[:3] if item.get("id")],
|
||
)
|
||
|
||
def _build_context_packet(
|
||
self,
|
||
truth_state: ProjectTruthStateProjection,
|
||
*,
|
||
body: ProjectCopilotAskRequest,
|
||
anchor: ProjectCopilotObjectAnchorPacket,
|
||
execution_summary: dict[str, Any],
|
||
) -> ProjectCopilotContextPacket:
|
||
pending_items = [item for item in truth_state.changeItems if str(item.get("status")) == "pending_review"]
|
||
accepted_or_edited = [
|
||
item for item in truth_state.changeItems if str(item.get("status")) in {"accepted", "edited"}
|
||
]
|
||
high_risk_items = [
|
||
item for item in truth_state.changeItems if _risk_level(item.get("riskLevel") or item.get("risk_level")) == "high"
|
||
]
|
||
selected_task = self._selected_task(truth_state, anchor)
|
||
active_level = self._action_ladder(body.surface, anchor, selected_task, len(pending_items))
|
||
allowed_actions = self._allowed_actions(body.surface, anchor, selected_task, len(pending_items), active_level)
|
||
forbidden_actions = self._forbidden_actions(body.surface, len(pending_items), selected_task)
|
||
limitations = self._limitations(truth_state, body.surface, anchor, selected_task, len(pending_items))
|
||
release_gate = execution_summary.get("releaseGate") if isinstance(execution_summary.get("releaseGate"), dict) else {}
|
||
release_signals = execution_summary.get("releaseSignals") if isinstance(execution_summary.get("releaseSignals"), dict) else {}
|
||
release_signal_summary = release_signals.get("summary") if isinstance(release_signals.get("summary"), dict) else {}
|
||
project_id = str(truth_state.project.id)
|
||
return ProjectCopilotContextPacket(
|
||
project_ref={
|
||
"id": project_id,
|
||
"title": truth_state.project.name,
|
||
"status": _stage_label(truth_state.project.stage),
|
||
"summary": truth_state.project.goal,
|
||
},
|
||
surface=body.surface,
|
||
object_anchor=anchor,
|
||
truth_projection={
|
||
"stage": _stage_label(truth_state.project.stage),
|
||
"loop_stage": _loop_stage_label(truth_state.loopStage),
|
||
"confirmed_requirements_count": len([item for item in truth_state.requirements if item.status == "confirmed"]),
|
||
"confirmed_tasks_count": len([item for item in truth_state.tasks if item.status in {"confirmed", "in_progress", "blocked", "done"}]),
|
||
"open_change_sets_count": len([item for item in truth_state.changeSets if str(item.get("status")) in {"open", "reviewed"}]),
|
||
"pending_change_items_count": len(pending_items),
|
||
"evidence_count": len(truth_state.evidence),
|
||
"high_risk_count": len(high_risk_items),
|
||
"latest_writeback_at": truth_state.lastWritebackAt,
|
||
},
|
||
change_context={
|
||
"active_change_set_id": truth_state.activeChangeSetId,
|
||
"selected_change_item_id": anchor.id if anchor.type == "change_item" else None,
|
||
"pending_count": len(pending_items),
|
||
"accepted_or_edited_count": len(accepted_or_edited),
|
||
"high_risk_count": len(high_risk_items),
|
||
"evidence_ids": list(dict.fromkeys(anchor.evidence_ids + self._pending_evidence_ids(pending_items)))[:6],
|
||
"commit_gate": "blocked_pending_items"
|
||
if pending_items
|
||
else "ready_for_human_commit"
|
||
if accepted_or_edited
|
||
else "no_open_changes",
|
||
},
|
||
execution_context={
|
||
"selected_task_id": str(selected_task.id) if selected_task else None,
|
||
"selected_task_status": selected_task.status if selected_task else None,
|
||
"formal_task_available": bool(selected_task and selected_task.status in {"confirmed", "in_progress", "blocked", "done"}),
|
||
"execution_status": truth_state.execution.status,
|
||
"prompt_package_ready": bool(truth_state.execution.prompt.strip()),
|
||
"repo_binding_known": bool(truth_state.execution.commitLog or truth_state.execution.prStatus != "pending"),
|
||
"artifact_completeness": _string_or_none(execution_summary.get("artifactCompleteness", {}).get("status") if isinstance(execution_summary.get("artifactCompleteness"), dict) else None),
|
||
"release_readiness": _string_or_none(release_gate.get("status")),
|
||
},
|
||
release_context={
|
||
"release_id": str(truth_state.releases[0].id) if truth_state.releases else None,
|
||
"release_status": _release_status_label(truth_state.releases[0].status) if truth_state.releases else None,
|
||
"release_gate_status": _string_or_none(release_gate.get("status")),
|
||
"feedback_signal_count": int(release_signal_summary.get("feedbackCount") or 0),
|
||
"metric_signal_count": len(truth_state.metricSignals),
|
||
"gate_blocked_count": len(release_gate.get("blockingReasons", []) or []),
|
||
},
|
||
allowed_actions=allowed_actions,
|
||
forbidden_actions=forbidden_actions,
|
||
automation_policy=_low_risk_automation_policy(forbidden_actions),
|
||
proactive_suggestions=self._proactive_suggestions(
|
||
project_id=project_id,
|
||
surface=body.surface,
|
||
anchor=anchor,
|
||
selected_task=selected_task,
|
||
pending_count=len(pending_items),
|
||
accepted_or_edited_count=len(accepted_or_edited),
|
||
release_gate=release_gate,
|
||
allowed_actions=allowed_actions,
|
||
),
|
||
limitations=limitations,
|
||
action_ladder_level=active_level,
|
||
)
|
||
|
||
def _render_answer(
|
||
self,
|
||
*,
|
||
body: ProjectCopilotAskRequest,
|
||
packet: ProjectCopilotContextPacket,
|
||
truth_state: ProjectTruthStateProjection,
|
||
) -> ProjectCopilotAnswerCanvas:
|
||
headline = f"{packet.object_anchor.title} · {self._intent_label(body.intent)}"
|
||
if body.intent == "supplement":
|
||
direct = (
|
||
f"补充信息会先挂到「{packet.object_anchor.title}」这个锚点下；如果它会改变项目事实，"
|
||
"下一步只能生成候选变更并进入待确认，不能直接写入项目记忆。"
|
||
)
|
||
elif body.intent == "command":
|
||
direct = (
|
||
f"这条指令只能在当前项目和「{packet.object_anchor.title}」范围内生成动作建议。"
|
||
"执行、发布、权限变更和项目记忆写入都必须回到对应主流程确认。"
|
||
)
|
||
else:
|
||
direct = self._explain_current_state(packet)
|
||
evidence_blocks = [
|
||
{
|
||
"display_label": str(item.get("title") or item.get("id")),
|
||
"display_summary": str(item.get("excerpt") or item.get("summary") or ""),
|
||
"business_values": {"source": str(item.get("sourceLabel") or item.get("source_label") or "项目证据")},
|
||
}
|
||
for item in truth_state.evidence
|
||
if str(item.get("id")) in set(packet.object_anchor.evidence_ids)
|
||
][:3]
|
||
next_actions = list(packet.proactive_suggestions[:5])
|
||
if not next_actions:
|
||
next_actions = [
|
||
{
|
||
"title": self._allowed_action_label(action),
|
||
"label": self._allowed_action_label(action),
|
||
"action": action,
|
||
"href": self._action_href(packet.surface, action, packet.project_ref["id"]),
|
||
"automation_level": "suggest_only",
|
||
"requires_user_confirm": True,
|
||
"object_ref": f"{packet.object_anchor.type}:{packet.object_anchor.id}",
|
||
"object_anchor": _canvas_object_anchor(packet),
|
||
}
|
||
for action in packet.allowed_actions
|
||
]
|
||
return ProjectCopilotAnswerCanvas(
|
||
headline=headline,
|
||
direct_answer=direct,
|
||
evidence_blocks=evidence_blocks,
|
||
object_refs=[
|
||
{
|
||
"object_ref": f"{packet.object_anchor.type}:{packet.object_anchor.id}",
|
||
"display_label": packet.object_anchor.title,
|
||
"display_summary": packet.object_anchor.summary,
|
||
}
|
||
],
|
||
next_actions=next_actions,
|
||
limitations=packet.limitations,
|
||
)
|
||
|
||
def _explain_current_state(self, packet: ProjectCopilotContextPacket) -> str:
|
||
pending = int(packet.change_context.get("pending_count") or 0)
|
||
high_risk = int(packet.change_context.get("high_risk_count") or 0)
|
||
if packet.surface == "change_review":
|
||
if pending:
|
||
return f"当前有 {pending} 个待确认变更项，其中高风险 {high_risk} 个。副驾只能解释证据、风险和处理顺序，最终写入必须由你确认。"
|
||
return "当前没有待确认变更项。副驾不会用示例对象补位，可以回到指挥台补充输入或查看已写入项目记忆。"
|
||
if packet.surface == "execution_bridge":
|
||
if packet.execution_context.get("formal_task_available"):
|
||
return "当前可以准备执行上下文包，但副驾不会在 PMFlow 内执行本机命令；真实执行必须由运行端领取正式执行包并回传证据。"
|
||
return "当前缺少已确认正式任务，候选任务不能直接进入执行桥。"
|
||
if packet.surface == "release_growth":
|
||
return "发布与反馈只能形成门禁判断或下一轮待确认变更，不能绕过项目记忆直接改正式对象。"
|
||
if pending:
|
||
return f"AI 已生成 {pending} 项待确认内容。下一步应核对证据、处理风险或进入待确认，而不是让副驾直接写入。"
|
||
return "当前副驾可以解释项目真相、补充候选方向或路由到主流程；没有人工确认前不会写入、执行或发布。"
|
||
|
||
def _action_ladder(
|
||
self,
|
||
surface: str,
|
||
anchor: ProjectCopilotObjectAnchorPacket,
|
||
selected_task: Any,
|
||
pending_count: int,
|
||
) -> ProjectCopilotActionLadderLevel:
|
||
if surface == "execution_bridge" or anchor.type == "task":
|
||
return "handoff_prepare"
|
||
if surface == "change_review" or anchor.risk_level == "high":
|
||
return "human_review"
|
||
if pending_count:
|
||
return "draft_candidate"
|
||
return "read_only"
|
||
|
||
def _allowed_actions(
|
||
self,
|
||
surface: str,
|
||
anchor: ProjectCopilotObjectAnchorPacket,
|
||
selected_task: Any,
|
||
pending_count: int,
|
||
ladder: ProjectCopilotActionLadderLevel,
|
||
) -> list[ProjectCopilotAllowedAction]:
|
||
actions: list[ProjectCopilotAllowedAction] = ["answer", "route_to_surface"]
|
||
if surface in {"command_center", "structured_extraction", "release_growth"} or ladder == "draft_candidate":
|
||
actions.append("draft_change")
|
||
if surface in {"change_review", "team_layer"} or pending_count > 0 or anchor.risk_level == "high":
|
||
actions.append("review_explain")
|
||
if (surface == "execution_bridge" or anchor.type == "task") and selected_task and selected_task.status in {"confirmed", "in_progress", "blocked", "done"}:
|
||
actions.append("prepare_execution_handoff")
|
||
return list(dict.fromkeys(actions))
|
||
|
||
def _forbidden_actions(self, surface: str, pending_count: int, selected_task: Any) -> list[dict[str, str]]:
|
||
forbidden = [
|
||
{"action": "cross_project_portfolio_ranking", "reason": "项目副驾只绑定当前项目，跨项目排序属于 Space AI。"},
|
||
{"action": "platform_provider_model_prompt_diagnosis", "reason": "provider/model/prompt 诊断只允许在平台 AI Ops 入口处理。"},
|
||
{"action": "project_memory_write", "reason": "项目记忆写入必须由 ChangeSet commit 主链执行。"},
|
||
{"action": "changeset_commit", "reason": "副驾只能解释、建议和路由，不能替用户提交变更集。"},
|
||
{"action": "external_command_execution", "reason": "外部命令必须由正式执行包和运行端确认链处理。"},
|
||
{"action": "publish_or_deploy", "reason": "发布、部署、删除和权限变更都必须经过发布门禁与人工确认。"},
|
||
]
|
||
if pending_count:
|
||
forbidden.append({"action": "execution_handoff_from_pending_change", "reason": "仍有待确认变更项，不能从候选对象直接进入执行桥。"})
|
||
if not selected_task or selected_task.status not in {"confirmed", "in_progress", "blocked", "done"}:
|
||
forbidden.append({"action": "prepare_execution_handoff", "reason": "缺少已确认正式任务，执行交接只能作为后续路由建议。"})
|
||
if surface == "team_layer":
|
||
forbidden.append({"action": "team_layer_direct_commit_or_assignment", "reason": "团队层只做责任路由和阻塞解释，不直接拍板写入或分派执行。"})
|
||
return forbidden
|
||
|
||
def _proactive_suggestions(
|
||
self,
|
||
*,
|
||
project_id: str,
|
||
surface: str,
|
||
anchor: ProjectCopilotObjectAnchorPacket,
|
||
selected_task: Any,
|
||
pending_count: int,
|
||
accepted_or_edited_count: int,
|
||
release_gate: dict[str, Any],
|
||
allowed_actions: list[ProjectCopilotAllowedAction],
|
||
) -> list[dict[str, Any]]:
|
||
suggestions: list[dict[str, Any]] = []
|
||
object_ref = f"{anchor.type}:{anchor.id}"
|
||
object_anchor = {
|
||
"object_ref": object_ref,
|
||
"object_type": anchor.type,
|
||
"object_id": anchor.id,
|
||
"surface": surface,
|
||
"project_id": project_id,
|
||
"project_title": anchor.title if anchor.type == "project" else "",
|
||
"href": self._surface_href(surface, project_id=project_id, anchor=anchor, selected_task=selected_task),
|
||
"parent_object_ref": None if anchor.type == "project" else f"project:{project_id}",
|
||
}
|
||
|
||
def push(
|
||
*,
|
||
title: str,
|
||
reason: str,
|
||
action: ProjectCopilotAllowedAction,
|
||
href: str,
|
||
automation_level: str = "suggest_only",
|
||
) -> None:
|
||
if action not in allowed_actions:
|
||
return
|
||
suggestions.append(
|
||
{
|
||
"title": title,
|
||
"label": title,
|
||
"reason": reason,
|
||
"why_now": reason,
|
||
"action": action,
|
||
"href": href,
|
||
"automation_level": automation_level,
|
||
"requires_user_confirm": True,
|
||
"required_confirmation": "human_confirm_on_target_surface",
|
||
"risk_level": anchor.risk_level,
|
||
"source_objects": [object_ref],
|
||
"audit_event": "project_copilot.proactive_suggestion.created",
|
||
"object_ref": object_ref,
|
||
"object_anchor": object_anchor,
|
||
"blocked_automation": [
|
||
"project_memory_write",
|
||
"changeset_commit",
|
||
"execution_handoff",
|
||
"external_command_execution",
|
||
"publish_or_deploy",
|
||
],
|
||
}
|
||
)
|
||
|
||
if pending_count > 0:
|
||
push(
|
||
title="进入待确认处理本轮变更",
|
||
reason=f"仍有 {pending_count} 个待确认项，AI 只能主动把你带到审查入口。",
|
||
action="review_explain",
|
||
href=f"/change-review?projectId={project_id}",
|
||
)
|
||
if accepted_or_edited_count > 0 and pending_count == 0:
|
||
push(
|
||
title="检查写入门禁再确认提交",
|
||
reason="已有人工处理过的候选项，但真正写入仍必须由你在 Change Review 提交。",
|
||
action="route_to_surface",
|
||
href=f"/change-review?projectId={project_id}",
|
||
)
|
||
if selected_task and str(getattr(selected_task, "status", "")) in {"confirmed", "in_progress", "blocked", "done"}:
|
||
push(
|
||
title="准备执行上下文包预览",
|
||
reason="只准备上下文和证据清单，不在 PMFlow 内执行本机命令。",
|
||
action="prepare_execution_handoff",
|
||
href=f"/execution-bridge?projectId={project_id}&taskId={getattr(selected_task, 'id', '')}",
|
||
automation_level="prepare_only",
|
||
)
|
||
if surface == "release_growth" or str(release_gate.get("status") or "").strip():
|
||
push(
|
||
title="检查发布门禁证据",
|
||
reason="发布只能先做证据检查和门禁解释，不能自动发布或部署。",
|
||
action="route_to_surface",
|
||
href=f"/release-growth?projectId={project_id}",
|
||
)
|
||
if not suggestions:
|
||
push(
|
||
title="继续补充项目上下文",
|
||
reason="AI 可以主动整理候选方向，但有效变更仍会进入待确认链。",
|
||
action="draft_change",
|
||
href=f"/command-center?projectId={project_id}",
|
||
)
|
||
return suggestions[:4]
|
||
|
||
def _surface_href(
|
||
self,
|
||
surface: str,
|
||
*,
|
||
project_id: str,
|
||
anchor: ProjectCopilotObjectAnchorPacket,
|
||
selected_task: Any,
|
||
) -> str:
|
||
if surface == "change_review":
|
||
if anchor.type == "change_item":
|
||
return f"/change-review?projectId={project_id}&itemId={anchor.id}"
|
||
if anchor.type == "change_set":
|
||
return f"/change-review?projectId={project_id}&changeSetId={anchor.id}"
|
||
return f"/change-review?projectId={project_id}"
|
||
if surface == "execution_bridge":
|
||
task_id = anchor.id if anchor.type == "task" else str(getattr(selected_task, "id", "") or "")
|
||
suffix = f"&taskId={task_id}" if task_id else ""
|
||
return f"/execution-bridge?projectId={project_id}{suffix}"
|
||
if surface == "release_growth":
|
||
suffix = f"&releaseId={anchor.id}" if anchor.type == "release" else ""
|
||
return f"/release-growth?projectId={project_id}{suffix}"
|
||
if surface == "team_layer":
|
||
return f"/team-layer?projectId={project_id}"
|
||
if surface == "project_memory":
|
||
return f"/project-memory?projectId={project_id}"
|
||
if surface == "work_graph":
|
||
return f"/work-graph?projectId={project_id}"
|
||
if surface == "structured_extraction":
|
||
return f"/structured-extraction?projectId={project_id}"
|
||
return f"/pmflow-os?projectId={project_id}"
|
||
|
||
def _limitations(
|
||
self,
|
||
truth_state: ProjectTruthStateProjection,
|
||
surface: str,
|
||
anchor: ProjectCopilotObjectAnchorPacket,
|
||
selected_task: Any,
|
||
pending_count: int,
|
||
) -> list[str]:
|
||
limitations: list[str] = []
|
||
if not truth_state.evidence:
|
||
limitations.append("当前项目缺少可引用证据，副驾不能给出已证实结论。")
|
||
if pending_count:
|
||
limitations.append(f"仍有 {pending_count} 个待确认变更项，正式写入前只能生成候选建议或审查解释。")
|
||
if (surface == "execution_bridge" or anchor.type == "task") and (
|
||
not selected_task or selected_task.status not in {"confirmed", "in_progress", "blocked", "done"}
|
||
):
|
||
limitations.append("缺少已确认正式任务，不能准备真实执行交接。")
|
||
if surface == "release_growth" and not truth_state.releases:
|
||
limitations.append("当前缺少发布对象和门禁记录，发布判断只能提示需要补齐证据。")
|
||
if anchor.mode in {"historical", "stale"}:
|
||
limitations.append("当前锚点不是实时对象，建议先切回当前变更、任务或项目全局。")
|
||
return limitations or ["所有建议仍需按当前页面主链确认后才会写入、执行或发布。"]
|
||
|
||
def _selected_task(self, truth_state: ProjectTruthStateProjection, anchor: ProjectCopilotObjectAnchorPacket) -> Any:
|
||
if anchor.type == "task":
|
||
return next((item for item in truth_state.tasks if str(item.id) == str(anchor.id)), None)
|
||
if truth_state.selectedTaskId:
|
||
selected = next((item for item in truth_state.tasks if str(item.id) == str(truth_state.selectedTaskId)), None)
|
||
if selected:
|
||
return selected
|
||
return truth_state.tasks[0] if truth_state.tasks else None
|
||
|
||
def _risk_for_change_set(self, truth_state: ProjectTruthStateProjection, change_set_id: str) -> str:
|
||
risks = [
|
||
_risk_level(item.get("riskLevel") or item.get("risk_level"))
|
||
for item in truth_state.changeItems
|
||
if str(item.get("changeSetId") or item.get("change_set_id")) == str(change_set_id)
|
||
]
|
||
if "high" in risks:
|
||
return "high"
|
||
if "medium" in risks:
|
||
return "medium"
|
||
return "low"
|
||
|
||
def _risk_for_stage(self, truth_state: ProjectTruthStateProjection, stage: str) -> str:
|
||
stage_risks = [item.level for item in truth_state.risks if item.stage == stage]
|
||
if "high" in stage_risks:
|
||
return "high"
|
||
if "medium" in stage_risks:
|
||
return "medium"
|
||
return "low"
|
||
|
||
def _evidence_for_change_set(self, truth_state: ProjectTruthStateProjection, change_set_id: str) -> list[str]:
|
||
evidence_ids: list[str] = []
|
||
for item in truth_state.changeItems:
|
||
if str(item.get("changeSetId") or item.get("change_set_id")) != str(change_set_id):
|
||
continue
|
||
evidence_ids.extend(str(eid) for eid in item.get("evidenceIds", []) or item.get("evidence_ids", []) or [])
|
||
return list(dict.fromkeys(evidence_ids))[:6]
|
||
|
||
def _pending_evidence_ids(self, pending_items: list[dict[str, Any]]) -> list[str]:
|
||
evidence_ids: list[str] = []
|
||
for item in pending_items:
|
||
evidence_ids.extend(str(eid) for eid in item.get("evidenceIds", []) or item.get("evidence_ids", []) or [])
|
||
return evidence_ids
|
||
|
||
def _intent_label(self, intent: str) -> str:
|
||
if intent == "supplement":
|
||
return "补充"
|
||
if intent == "command":
|
||
return "指令"
|
||
return "追问"
|
||
|
||
def _allowed_action_label(self, action: str) -> str:
|
||
return {
|
||
"answer": "回答当前项目",
|
||
"draft_change": "生成候选变更",
|
||
"review_explain": "解释审查风险",
|
||
"prepare_execution_handoff": "准备执行交接",
|
||
"route_to_surface": "跳转主流程",
|
||
}.get(action, "继续处理")
|
||
|
||
def _action_href(self, surface: str, action: str, project_id: str) -> str:
|
||
if action == "review_explain":
|
||
return f"/change-review?projectId={project_id}"
|
||
if action == "prepare_execution_handoff":
|
||
return f"/execution-bridge?projectId={project_id}"
|
||
if action == "draft_change":
|
||
return f"/structured-extraction?projectId={project_id}"
|
||
return f"/{surface.replace('_', '-')}?projectId={project_id}"
|
||
|
||
|
||
def _find_by_id(items: list[dict[str, Any]], item_id: str) -> dict[str, Any] | None:
|
||
return next((item for item in items if str(item.get("id")) == str(item_id)), None)
|
||
|
||
|
||
def _low_risk_automation_policy(forbidden_actions: list[dict[str, str]]) -> dict[str, Any]:
|
||
return {
|
||
"policy_version": "ai_dual_04_low_risk_proactive.v1",
|
||
"mode": "suggest_only",
|
||
"allowed_automation_levels": ["suggest_only", "prepare_only"],
|
||
"allowed_without_user_confirmation": [
|
||
"rank_attention",
|
||
"prepare_navigation",
|
||
"summarize_context",
|
||
"prepare_execution_context_preview",
|
||
],
|
||
"requires_user_confirmation_before": [
|
||
"project_memory_write",
|
||
"changeset_commit",
|
||
"execution_handoff",
|
||
"external_command_execution",
|
||
"publish_or_deploy",
|
||
],
|
||
"forbidden_actions": [item.get("action") for item in forbidden_actions if item.get("action")],
|
||
}
|
||
|
||
|
||
def _canvas_object_anchor(packet: ProjectCopilotContextPacket) -> dict[str, Any]:
|
||
return {
|
||
"object_ref": f"{packet.object_anchor.type}:{packet.object_anchor.id}",
|
||
"object_type": packet.object_anchor.type,
|
||
"object_id": packet.object_anchor.id,
|
||
"surface": packet.surface,
|
||
"project_id": str(packet.project_ref.get("id") or ""),
|
||
"project_title": str(packet.project_ref.get("title") or ""),
|
||
"href": f"/{str(packet.surface).replace('_', '-')}?projectId={packet.project_ref.get('id')}",
|
||
"parent_object_ref": None if packet.object_anchor.type == "project" else f"project:{packet.project_ref.get('id')}",
|
||
}
|
||
|
||
|
||
def _risk_level(value: Any) -> str:
|
||
normalized = str(value or "low").lower()
|
||
if normalized == "high":
|
||
return "high"
|
||
if normalized == "medium":
|
||
return "medium"
|
||
return "low"
|
||
|
||
|
||
def _stage_label(value: str) -> str:
|
||
return {
|
||
"idea": "想法",
|
||
"validation": "验证",
|
||
"planning": "规划",
|
||
"development": "开发",
|
||
"testing": "测试",
|
||
"release": "发布",
|
||
"growth": "增长",
|
||
}.get(str(value), str(value))
|
||
|
||
|
||
def _loop_stage_label(value: str) -> str:
|
||
return {
|
||
"command_idle": "等待输入",
|
||
"ai_extracting": "AI 提取中",
|
||
"extraction_ready": "提取待核对",
|
||
"review_open": "待确认",
|
||
"writeback_success": "写入完成",
|
||
}.get(str(value), str(value))
|
||
|
||
|
||
def _change_set_status_label(value: Any) -> str:
|
||
return {
|
||
"committed": "已写入",
|
||
"partially_committed": "部分写入",
|
||
"reviewed": "已审查",
|
||
"open": "待确认",
|
||
}.get(str(value), "待确认")
|
||
|
||
|
||
def _change_item_status_label(value: Any) -> str:
|
||
return {
|
||
"accepted": "已接受",
|
||
"edited": "已修改",
|
||
"deferred": "暂不落库",
|
||
"committed": "已写入",
|
||
"skipped": "已跳过",
|
||
"pending_review": "待确认",
|
||
}.get(str(value), "待确认")
|
||
|
||
|
||
def _task_status_label(value: str) -> str:
|
||
return {
|
||
"blocked": "阻塞",
|
||
"done": "完成",
|
||
"in_progress": "执行中",
|
||
"confirmed": "正式任务",
|
||
"draft": "草稿",
|
||
}.get(str(value), str(value))
|
||
|
||
|
||
def _release_status_label(value: str) -> str:
|
||
return {
|
||
"ready": "可发布",
|
||
"shipping": "发布中",
|
||
"live": "已发布",
|
||
"watch": "观察中",
|
||
"check_pending": "待检查",
|
||
}.get(str(value), str(value))
|
||
|
||
|
||
def _string_or_none(value: Any) -> str | None:
|
||
if value is None:
|
||
return None
|
||
text = str(value).strip()
|
||
return text or None
|