From b87951935a22cf7be590e8ac3b3fec1206d5a5ce Mon Sep 17 00:00:00 2001 From: Albert Date: Sun, 26 Apr 2026 09:14:56 +0800 Subject: [PATCH] feat: implement PMFlow changeset commit backend --- backend/app/api/__init__.py | 56 +-- backend/app/api/pmflow_os_api/__init__.py | 25 ++ backend/app/schemas/pmflow_os.py | 8 +- .../services/pmflow_change_review_service.py | 391 +++++++++++++++++- .../tests/test_pmflow_change_review_api.py | 132 ++++++ docs/PMFLOW_PRODUCT_DELIVERY_STANDARD.md | 44 +- docs/architecture/CODE_REFERENCE_INDEX.md | 2 +- .../PMFLOW_OS_PRODUCT_CONTRACT.md | 11 +- 8 files changed, 608 insertions(+), 61 deletions(-) diff --git a/backend/app/api/__init__.py b/backend/app/api/__init__.py index 8de3e614..1963a3b9 100644 --- a/backend/app/api/__init__.py +++ b/backend/app/api/__init__.py @@ -1,55 +1,11 @@ -""" -API 路由模块 +"""API package. -按业务域拆分路由，便于维护与扩展。 +Route registration lives in ``app.api.router``. Keep this package entrypoint +lightweight so focused tests can import a single API module without importing +optional integrations such as Feishu or MCP. """ -from .signals_api import router as signals_router -from .feedbacks import router as feedbacks_router -from .experiments import router as experiments_router -from .context import router as context_router -from .tasks import router as tasks_router -from .webhooks import router as webhooks_router -from .auth import router as auth_router -from .users_api import router as users_router -from .endpoints.skill import router as skill_router -from .sstc_api import router as sstc_router -from .ai_steward_api import router as ai_steward_router -from .feishu_api import router as feishu_router -from .wecom_api import router as wecom_router -from .model_capability_api import router as model_capability_router -from .budget import router as budget_router -from .scheduler import router as scheduler_router -from .notifications import router as notifications_router -from .llm_engines import router as llm_engines_router -from .llm_providers import router as llm_providers_router -from .prompt_profiles import router as prompt_profiles_router -from .knowledge_graph import router as knowledge_graph_router -from .operation_catalog import router as operation_catalog_router -from .ai_pmo_api import router as ai_pmo_router - __all__ = [ - "signals_router", - "feedbacks_router", - "experiments_router", - "context_router", - "tasks_router", - "webhooks_router", - "auth_router", - "users_router", - "skill_router", - "sstc_router", - "ai_steward_router", - "feishu_router", - "wecom_router", - "model_capability_router", - "budget_router", - "scheduler_router", - "notifications_router", - "llm_engines_router", - "llm_providers_router", - "prompt_profiles_router", - "knowledge_graph_router", - "operation_catalog_router", - "ai_pmo_router", + "pmflow_launch_api", + "pmflow_os_api", ] diff --git a/backend/app/api/pmflow_os_api/__init__.py b/backend/app/api/pmflow_os_api/__init__.py index 72c0703c..fcf674b5 100644 --- a/backend/app/api/pmflow_os_api/__init__.py +++ b/backend/app/api/pmflow_os_api/__init__.py @@ -12,6 +12,7 @@ from ...models import Project, User, UserProject from ...schemas.pmflow_os import ( PMFlowChangeItemDeferRequest, PMFlowChangeItemEditRequest, + PMFlowChangeSetCommitRequest, PMFlowExtractionCreateRequest, PMFlowThreadMessageCreateRequest, ) @@ -199,6 +200,30 @@ async def defer_pmflow_os_change_item( return success_response(payload) +@router.post("/change-sets/{change_set_id}/commit") +async def commit_pmflow_os_change_set( + change_set_id: int, + body: PMFlowChangeSetCommitRequest | None = None, + db: AsyncSession = Depends(get_db), + current_user: User = Depends(get_current_user), +): + tenant_id = int(current_user.tenant_id) + service = PMFlowChangeReviewService(db, tenant_id=tenant_id) + project_id = await service.get_change_set_project_id(change_set_id=int(change_set_id)) + if project_id is None: + return JSONResponse(status_code=404, content=error_response(404, "changeset_not_found")) + await _ensure_project_member_or_manager(db, tenant_id=tenant_id, project_id=project_id, current_user=current_user) + try: + payload = await service.commit_change_set( + change_set_id=int(change_set_id), + body=body or PMFlowChangeSetCommitRequest(), + user_id=int(current_user.id), + ) + except PMFlowChangeReviewError as exc: + return _review_error(exc) + return success_response(payload) + + @router.get("/projects/{project_id}/execution-summary") async def get_pmflow_os_execution_summary( project_id: int, diff --git a/backend/app/schemas/pmflow_os.py b/backend/app/schemas/pmflow_os.py index 41546ff2..70e2a145 100644 --- a/backend/app/schemas/pmflow_os.py +++ b/backend/app/schemas/pmflow_os.py @@ -14,7 +14,7 @@ PMFlowExecutionStatus = Literal["queued", "pushed", "coding", "pr_open", "testin PMFlowReleaseStatus = Literal["check_pending", "ready", "shipping", "live", "watch"] ProjectLoopStage = Literal["command_idle", "ai_extracting", "extraction_ready", "review_open", "writeback_success"] PMFlowChangeGroup = Literal["new_object", "field_update", "risk", "decision"] -PMFlowChangeStatus = Literal["pending_review", "accepted", "edited", "deferred"] +PMFlowChangeStatus = Literal["pending_review", "accepted", "edited", "deferred", "committed", "skipped"] PMFlowObjectType = Literal[ "Requirement", "Decision", @@ -80,6 +80,12 @@ class PMFlowChangeItemDeferRequest(BaseModel): reason: Optional[str] = None +class PMFlowChangeSetCommitRequest(BaseModel): + item_ids: list[str] = Field(default_factory=list) + mode: Literal["atomic"] = "atomic" + note: Optional[str] = None + + class PMFlowChangeSetDetailResponse(BaseModel): changeSet: dict[str, Any] items: list[dict[str, Any]] = Field(default_factory=list) diff --git a/backend/app/services/pmflow_change_review_service.py b/backend/app/services/pmflow_change_review_service.py index cdf801f4..f5ff54f5 100644 --- a/backend/app/services/pmflow_change_review_service.py +++ b/backend/app/services/pmflow_change_review_service.py @@ -14,9 +14,13 @@ from ..models import ( PMFlowEvidenceLink, PMFlowThread, PMFlowThreadMessage, + SemanticFact, + Task, + TruthNode, ) from ..schemas.pmflow_os import ( PMFlowChangeItemEditRequest, + PMFlowChangeSetCommitRequest, PMFlowExtractionCreateRequest, PMFlowThreadMessageCreateRequest, ) @@ -32,6 +36,8 @@ class PMFlowChangeReviewError(RuntimeError): class PMFlowChangeReviewService: + COMMIT_OBJECT_TYPES = {"Requirement", "Decision", "Task", "Risk"} + def __init__(self, db: AsyncSession, *, tenant_id: int): self.db = db self.tenant_id = int(tenant_id) @@ -155,6 +161,15 @@ class PMFlowChangeReviewService: value = result.scalar_one_or_none() return int(value) if value is not None else None + async def get_change_set_project_id(self, *, change_set_id: int) -> Optional[int]: + result = await self.db.execute( + select(PMFlowChangeSet.project_id).where( + and_(PMFlowChangeSet.tenant_id == self.tenant_id, PMFlowChangeSet.id == int(change_set_id)) + ) + ) + value = result.scalar_one_or_none() + return int(value) if value is not None else None + async def accept_item(self, *, change_item_id: int, user_id: int) -> dict[str, Any]: item, change_set = await self._load_item_with_set(change_item_id=change_item_id) await self._review_item(item=item, change_set=change_set, action="accept", decision="approved", user_id=user_id) @@ -198,6 +213,84 @@ class PMFlowChangeReviewService: await self.db.commit() return await self.get_change_set_detail(project_id=int(item.project_id), change_set_id=int(item.change_set_id)) + async def commit_change_set( + self, + *, + change_set_id: int, + body: PMFlowChangeSetCommitRequest, + user_id: int, + ) -> dict[str, Any]: + try: + change_set = await self._load_change_set_for_commit(change_set_id=change_set_id) + items = await self._load_items(project_id=int(change_set.project_id), change_set_id=int(change_set.id)) + evidence = await self._load_evidence(project_id=int(change_set.project_id), change_set_id=int(change_set.id)) + approvals = await self._load_approvals(project_id=int(change_set.project_id), change_set_id=int(change_set.id)) + selected_ids = self._parse_commit_item_ids(body.item_ids) + self._validate_commit_preflight(change_set=change_set, items=items, selected_ids=selected_ids) + eligible_items = [ + item + for item in items + if item.status in {"accepted", "edited"} and (not selected_ids or int(item.id) in selected_ids) + ] + if not eligible_items: + raise PMFlowChangeReviewError("no_accepted_items", "no_accepted_items", status_code=409) + + committed_items: list[dict[str, Any]] = [] + for item in eligible_items: + self._validate_commit_item(item=item, evidence=evidence, approvals=approvals) + for item in eligible_items: + committed_items.append( + await self._commit_item( + item=item, + change_set=change_set, + evidence=evidence, + user_id=user_id, + ) + ) + + now = now_cn_naive() + for item in eligible_items: + item.status = "committed" + remaining_open = [ + item + for item in items + if item.id not in {committed["changeItemIdRaw"] for committed in committed_items} + and item.status in {"pending_review", "accepted", "edited"} + ] + if not remaining_open: + change_set.status = "committed" + change_set.committed_at = now + await self.db.commit() + + skipped_items = [ + self._serialize_skipped_commit_item(item) + for item in items + if item.id not in {committed["changeItemIdRaw"] for committed in committed_items} + and item.status in {"deferred", "skipped"} + ] + return { + "changeSet": { + "id": str(change_set.id), + "status": change_set.status, + "rawStatus": change_set.status, + "committedAt": change_set.committed_at.isoformat() if change_set.committed_at else None, + "committedItemIds": [item["changeItemId"] for item in committed_items], + "skippedItemIds": [item["changeItemId"] for item in skipped_items], + "failedItemIds": [], + }, + "committedItems": [ + {key: value for key, value in item.items() if key != "changeItemIdRaw"} for item in committed_items + ], + "skippedItems": skipped_items, + "failedItems": [], + } + except PMFlowChangeReviewError: + await self.db.rollback() + raise + except Exception as exc: + await self.db.rollback() + raise PMFlowChangeReviewError("commit_failed", "commit_failed", status_code=500) from exc + async def _resolve_or_create_thread(self, *, project_id: int, thread_id: str) -> PMFlowThread: cleaned = str(thread_id or "").strip() if cleaned.isdigit(): @@ -276,6 +369,17 @@ class PMFlowChangeReviewService: raise PMFlowChangeReviewError("changeset_not_found", "changeset_not_found", status_code=404) return row + async def _load_change_set_for_commit(self, *, change_set_id: int) -> PMFlowChangeSet: + result = await self.db.execute( + select(PMFlowChangeSet) + .where(and_(PMFlowChangeSet.tenant_id == self.tenant_id, PMFlowChangeSet.id == int(change_set_id))) + .with_for_update() + ) + row = result.scalar_one_or_none() + if row is None: + raise PMFlowChangeReviewError("changeset_not_found", "changeset_not_found", status_code=404) + return row + async def _load_items(self, *, project_id: int, change_set_id: int) -> list[PMFlowChangeItem]: result = await self.db.execute( select(PMFlowChangeItem) @@ -381,6 +485,287 @@ class PMFlowChangeReviewService: change_set.status = "reviewed" change_set.reviewed_at = now_cn_naive() + @staticmethod + def _parse_commit_item_ids(raw_item_ids: list[str]) -> set[int]: + parsed: set[int] = set() + for raw in raw_item_ids or []: + text = str(raw or "").strip() + if not text.isdigit(): + raise PMFlowChangeReviewError("invalid_commit_item_ids", "invalid_commit_item_ids", status_code=422) + parsed.add(int(text)) + return parsed + + def _validate_commit_preflight( + self, + *, + change_set: PMFlowChangeSet, + items: list[PMFlowChangeItem], + selected_ids: set[int], + ) -> None: + if change_set.status == "committed": + raise PMFlowChangeReviewError("changeset_already_committed", "changeset_already_committed", status_code=409) + if change_set.status not in {"open", "reviewed"}: + raise PMFlowChangeReviewError("changeset_not_reviewed", "changeset_not_reviewed", status_code=409) + item_ids = {int(item.id) for item in items} + if selected_ids and not selected_ids.issubset(item_ids): + raise PMFlowChangeReviewError("invalid_commit_item_ids", "invalid_commit_item_ids", status_code=422) + if selected_ids: + invalid_selected = [ + item + for item in items + if int(item.id) in selected_ids and item.status not in {"accepted", "edited"} + ] + if invalid_selected: + raise PMFlowChangeReviewError("invalid_commit_item_ids", "invalid_commit_item_ids", status_code=422) + return + if any(item.status == "pending_review" for item in items): + raise PMFlowChangeReviewError("changeset_not_reviewed", "changeset_not_reviewed", status_code=409) + + def _validate_commit_item( + self, + *, + item: PMFlowChangeItem, + evidence: list[PMFlowEvidenceLink], + approvals: list[PMFlowApproval], + ) -> None: + if item.object_type not in self.COMMIT_OBJECT_TYPES: + raise PMFlowChangeReviewError("unsupported_object_type", "unsupported_object_type", status_code=422) + payload = self._effective_payload(item) + self._require_commit_text(item.title, field="title") + self._require_commit_text(str(payload.get("summary") or item.summary), field="summary") + if item.object_type == "Risk": + self._require_commit_text(str(payload.get("mitigation") or payload.get("mitigation_plan") or item.summary), field="mitigation") + if item.object_type == "Task": + self._require_commit_text(str(payload.get("description") or payload.get("summary") or item.summary), field="description") + if str(item.risk_level or "").lower() == "high": + approval_ok = any( + int(row.change_item_id) == int(item.id) + and row.decision in {"approved", "edited"} + and row.after_status in {"accepted", "edited"} + for row in approvals + ) + if not approval_ok or not self._item_evidence(item=item, evidence=evidence): + raise PMFlowChangeReviewError("high_risk_approval_required", "high_risk_approval_required", status_code=409) + + async def _commit_item( + self, + *, + item: PMFlowChangeItem, + change_set: PMFlowChangeSet, + evidence: list[PMFlowEvidenceLink], + user_id: int, + ) -> dict[str, Any]: + if item.object_type == "Task": + target_type, target_id = await self._commit_task_item( + item=item, + change_set=change_set, + evidence=evidence, + user_id=user_id, + ) + else: + target_type, target_id = await self._commit_truth_node_item( + item=item, + change_set=change_set, + evidence=evidence, + user_id=user_id, + ) + item.object_id = str(target_id) + evidence_ids = [str(row.id) for row in self._item_evidence(item=item, evidence=evidence)] + return { + "changeItemIdRaw": int(item.id), + "changeItemId": str(item.id), + "objectType": item.object_type, + "targetType": target_type, + "targetId": str(target_id), + "status": "committed", + "evidenceIds": evidence_ids, + "semanticFactStatus": "seeded", + } + + async def _commit_truth_node_item( + self, + *, + item: PMFlowChangeItem, + change_set: PMFlowChangeSet, + evidence: list[PMFlowEvidenceLink], + user_id: int, + ) -> tuple[str, int]: + node_type = { + "Requirement": "requirement", + "Decision": "decision", + "Risk": "risk", + }.get(item.object_type) + if node_type is None: + raise PMFlowChangeReviewError("unsupported_object_type", "unsupported_object_type", status_code=422) + content = self._commit_content(item=item, change_set=change_set, evidence=evidence, user_id=user_id) + node = TruthNode( + tenant_id=self.tenant_id, + node_type=node_type, + project_id=int(item.project_id), + signal_id=None, + created_by=int(user_id), + title=item.title[:500], + content=content, + version=1, + status="confirmed", + ) + self.db.add(node) + await self.db.flush() + self._add_semantic_fact_for_commit(node=node, item=item, evidence=evidence) + return "truth_node", int(node.id) + + async def _commit_task_item( + self, + *, + item: PMFlowChangeItem, + change_set: PMFlowChangeSet, + evidence: list[PMFlowEvidenceLink], + user_id: int, + ) -> tuple[str, int]: + content = self._commit_content(item=item, change_set=change_set, evidence=evidence, user_id=user_id) + node = TruthNode( + tenant_id=self.tenant_id, + node_type="task", + project_id=int(item.project_id), + signal_id=None, + created_by=int(user_id), + title=item.title[:500], + content=content, + version=1, + status="confirmed", + ) + self.db.add(node) + await self.db.flush() + payload = self._effective_payload(item) + task = Task( + tenant_id=self.tenant_id, + truth_node_id=int(node.id), + project_id=int(item.project_id), + assignee_id=None, + ai_recommended_assignees=None, + ai_breakdown_payload={"source": "pmflow_change_set_commit", "payload": payload}, + manual_overrides={ + "change_set_id": int(change_set.id), + "change_item_id": int(item.id), + "acceptance_criteria": self._acceptance_criteria(payload), + }, + title=item.title[:500], + description=str(payload.get("description") or payload.get("summary") or item.summary), + required_skill_tag_ids=[], + status="todo", + priority=self._priority(payload.get("priority")), + ) + self.db.add(task) + await self.db.flush() + self._add_semantic_fact_for_commit(node=node, item=item, evidence=evidence) + self._add_semantic_fact_for_commit(node=None, item=item, evidence=evidence, task=task) + return "task", int(task.id) + + def _commit_content( + self, + *, + item: PMFlowChangeItem, + change_set: PMFlowChangeSet, + evidence: list[PMFlowEvidenceLink], + user_id: int, + ) -> dict[str, Any]: + payload = self._effective_payload(item) + evidence_ids = [str(row.id) for row in self._item_evidence(item=item, evidence=evidence)] + return { + "source": "pmflow_change_set_commit", + "change_set_id": int(change_set.id), + "change_item_id": int(item.id), + "object_type": item.object_type, + "summary": str(payload.get("summary") or item.summary), + "diff": item.diff, + "payload": payload, + "stage": str(payload.get("stage") or ("development" if item.object_type == "Task" else "planning")), + "owner": str(payload.get("owner") or payload.get("decision_owner") or "未分配"), + "mitigation": str(payload.get("mitigation") or payload.get("mitigation_plan") or item.summary), + "evidence_ids": evidence_ids, + "confidence": float(item.confidence or 0.0), + "risk_level": item.risk_level, + "committed_by": int(user_id), + "committed_at": now_cn_naive().isoformat(), + } + + def _add_semantic_fact_for_commit( + self, + *, + node: Optional[TruthNode], + item: PMFlowChangeItem, + evidence: list[PMFlowEvidenceLink], + task: Optional[Task] = None, + ) -> None: + target_type = "task" if task is not None else "truth_node" + target_id = int(task.id) if task is not None else int(node.id) # type: ignore[union-attr] + source_evidence = self._item_evidence(item=item, evidence=evidence) + first_evidence = source_evidence[0] if source_evidence else None + text = str(self._effective_payload(item).get("summary") or item.summary).strip() + if not text: + return + self.db.add( + SemanticFact( + tenant_id=self.tenant_id, + entity_type=target_type, + entity_id=target_id, + fact_namespace="pmflow_change_commit", + fact_key="summary", + fact_value_text=text, + fact_value_norm=text[:255], + value_type="text", + confidence=float(item.confidence or 0.0), + evidence_source_type="pmflow_evidence_link" if first_evidence is not None else None, + evidence_source_id=int(first_evidence.id) if first_evidence is not None else None, + evidence_excerpt=first_evidence.excerpt if first_evidence is not None else None, + extract_version="pmflow_change_commit_v1", + freshness_at=now_cn_naive().isoformat(), + is_active=True, + ) + ) + + @staticmethod + def _effective_payload(item: PMFlowChangeItem) -> dict[str, Any]: + payload = item.edited_payload if item.status == "edited" and item.edited_payload is not None else item.payload + return dict(payload or {}) + + @staticmethod + def _item_evidence(*, item: PMFlowChangeItem, evidence: list[PMFlowEvidenceLink]) -> list[PMFlowEvidenceLink]: + return [row for row in evidence if row.change_item_id in {None, item.id}] + + @staticmethod + def _require_commit_text(value: str, *, field: str) -> None: + if not str(value or "").strip(): + raise PMFlowChangeReviewError("invalid_change_payload", f"missing_{field}", status_code=422) + + @staticmethod + def _priority(value: Any) -> str: + text = str(value or "medium").strip().lower() + if text in {"high", "medium", "low"}: + return text + if text in {"p0", "critical", "urgent"}: + return "high" + if text in {"p2", "p3"}: + return "low" + return "medium" + + @staticmethod + def _acceptance_criteria(payload: dict[str, Any]) -> list[str]: + raw = payload.get("acceptanceCriteria") or payload.get("acceptance_criteria") or [] + if isinstance(raw, list): + return [str(item).strip() for item in raw if str(item).strip()] + text = str(raw or "").strip() + return [text] if text else [] + + @staticmethod + def _serialize_skipped_commit_item(item: PMFlowChangeItem) -> dict[str, Any]: + return { + "changeItemId": str(item.id), + "objectType": item.object_type, + "status": item.status, + "reason": "item_not_accepted_or_edited", + } + @staticmethod def serialize_message(row: PMFlowThreadMessage) -> dict[str, Any]: return { @@ -398,16 +783,18 @@ class PMFlowChangeReviewService: @staticmethod def serialize_change_set(row: PMFlowChangeSet, items: list[PMFlowChangeItem]) -> dict[str, Any]: + status = row.status if row.status in {"open", "reviewed", "committed", "partially_committed"} else "reviewed" return { "id": str(row.id), "threadId": str(row.thread_id), "summary": row.summary, "headline": row.headline, "createdAt": row.created_at.isoformat() if row.created_at else "", - "status": "open" if row.status == "open" else "reviewed", + "status": status, "itemIds": [str(item.id) for item in items], "sourceMessageId": str(row.source_message_id), "rawStatus": row.status, + "committedAt": row.committed_at.isoformat() if row.committed_at else None, } @staticmethod @@ -424,7 +811,7 @@ class PMFlowChangeReviewService: "diff": row.diff, "confidence": float(row.confidence or 0.0), "riskLevel": row.risk_level, - "status": row.status if row.status in {"pending_review", "accepted", "edited", "deferred"} else "deferred", + "status": row.status if row.status in {"pending_review", "accepted", "edited", "deferred", "committed", "skipped"} else "deferred", "evidenceIds": evidence_ids, "payload": row.payload or {}, "editedPayload": row.edited_payload, diff --git a/backend/tests/test_pmflow_change_review_api.py b/backend/tests/test_pmflow_change_review_api.py index ff48cd72..a7eecaa1 100644 --- a/backend/tests/test_pmflow_change_review_api.py +++ b/backend/tests/test_pmflow_change_review_api.py @@ -182,6 +182,46 @@ def _proposed_items(): ] +def _commit_proposed_items(): + return [ + { + "group": "new_object", + "object_type": "Requirement", + "title": "新增需求：真实写入项目记忆", + "summary": "确认后的需求必须进入 truth_nodes。", + "diff": "写入 requirement truth node。", + "payload": {"summary": "确认后的需求", "stage": "planning"}, + "confidence": 0.92, + "risk_level": "medium", + }, + { + "group": "new_object", + "object_type": "Task", + "title": "新增任务：生成开发上下文包", + "summary": "确认后的任务必须进入 tasks。", + "diff": "写入 task。", + "payload": { + "summary": "确认后的任务", + "description": "生成开发上下文包并等待执行桥消费。", + "priority": "high", + "acceptanceCriteria": ["任务进入 tasks", "任务绑定 truth_node"], + }, + "confidence": 0.88, + "risk_level": "medium", + }, + { + "group": "risk", + "object_type": "Risk", + "title": "新增风险：commit 需要审批", + "summary": "高风险项必须有 approval 才能写入。", + "diff": "写入 high risk。", + "payload": {"summary": "确认后的高风险", "mitigation": "提交前检查 approval gate。"}, + "confidence": 0.81, + "risk_level": "high", + }, + ] + + async def _create_message_and_extraction(client, project_id: int): message = await client.post( f"/pmflow-os/projects/{project_id}/threads/thread-project-{project_id}/messages", @@ -204,6 +244,28 @@ async def _create_message_and_extraction(client, project_id: int): return msg, extraction.json()["data"] +async def _create_commit_extraction(client, project_id: int): + message = await client.post( + f"/pmflow-os/projects/{project_id}/threads/thread-project-{project_id}/messages", + json={"content": "请把本轮确认项写入项目真相层", "source": "web", "attachments": []}, + ) + assert message.status_code == 200 + msg = message.json()["data"] + extraction = await client.post( + f"/pmflow-os/projects/{project_id}/extractions", + json={ + "thread_id": msg["threadId"], + "message_id": msg["id"], + "scope": "latest_message", + "headline": "待提交变更", + "summary": "生成 3 个可提交候选。", + "proposed_items": _commit_proposed_items(), + }, + ) + assert extraction.status_code == 200 + return msg, extraction.json()["data"] + + @pytest.mark.asyncio async def test_change_review_message_extraction_detail_and_projection(db_env) -> None: session = db_env @@ -269,6 +331,76 @@ async def test_change_item_accept_edit_defer_write_approvals_and_do_not_write_me assert len(truth.json()["data"]["approvals"]) == 3 +@pytest.mark.asyncio +async def test_change_set_commit_writes_truth_nodes_tasks_and_semantic_facts(db_env) -> None: + session = db_env + user, project = await _seed_user_project(session) + app = _build_app(session, SimpleNamespace(id=user.id, tenant_id=1, role="MEMBER")) + + async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client: + _, detail = await _create_commit_extraction(client, project.id) + item_ids = [item["id"] for item in detail["items"]] + await client.post(f"/pmflow-os/change-items/{item_ids[0]}/accept") + await client.post(f"/pmflow-os/change-items/{item_ids[1]}/accept") + await client.post( + f"/pmflow-os/change-items/{item_ids[2]}/edit", + json={"summary": "人工确认后的高风险", "payload_patch": {"mitigation": "人工确认后再落库"}}, + ) + committed = await client.post( + f"/pmflow-os/change-sets/{detail['changeSet']['id']}/commit", + json={"mode": "atomic", "note": "确认写入项目记忆"}, + ) + truth = await client.get(f"/pmflow-os/projects/{project.id}/truth-state") + + assert committed.status_code == 200 + payload = committed.json()["data"] + assert payload["changeSet"]["status"] == "committed" + assert len(payload["committedItems"]) == 3 + assert payload["failedItems"] == [] + assert await session.scalar(select(func.count(TruthNode.id))) == 3 + assert await session.scalar(select(func.count(Task.id))) == 1 + assert await session.scalar(select(func.count(SemanticFact.id))) == 4 + + task = (await session.execute(select(Task))).scalar_one() + assert task.priority == "high" + assert task.status == "todo" + assert task.manual_overrides["acceptance_criteria"] == ["任务进入 tasks", "任务绑定 truth_node"] + + truth_payload = truth.json()["data"] + assert truth_payload["activeChangeSetId"] is None + assert truth_payload["projectMemory"]["sourceSummary"]["truth_nodes"] == 3 + assert truth_payload["projectMemory"]["sourceSummary"]["semantic_facts"] == 4 + + +@pytest.mark.asyncio +async def test_change_set_commit_blocks_high_risk_without_approval_and_rolls_back(db_env) -> None: + session = db_env + user, project = await _seed_user_project(session) + app = _build_app(session, SimpleNamespace(id=user.id, tenant_id=1, role="MEMBER")) + + async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client: + _, detail = await _create_commit_extraction(client, project.id) + high_risk_id = detail["items"][2]["id"] + + item = (await session.execute(select(PMFlowChangeItem).where(PMFlowChangeItem.id == int(high_risk_id)))).scalar_one() + item.status = "accepted" + change_set = (await session.execute(select(PMFlowChangeSet).where(PMFlowChangeSet.id == int(item.change_set_id)))).scalar_one() + change_set.status = "reviewed" + await session.commit() + + async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client: + blocked = await client.post( + f"/pmflow-os/change-sets/{detail['changeSet']['id']}/commit", + json={"item_ids": [high_risk_id], "mode": "atomic"}, + ) + + assert blocked.status_code == 409 + assert "high_risk_approval_required" in blocked.json()["message"] + assert await session.scalar(select(func.count(TruthNode.id))) == 0 + assert await session.scalar(select(func.count(Task.id))) == 0 + assert await session.scalar(select(func.count(SemanticFact.id))) == 0 + + @pytest.mark.asyncio async def test_change_item_review_conflict_and_closed_changeset_guard(db_env) -> None: session = db_env diff --git a/docs/PMFLOW_PRODUCT_DELIVERY_STANDARD.md b/docs/PMFLOW_PRODUCT_DELIVERY_STANDARD.md index d3f1a185..f8fc1a82 100644 --- a/docs/PMFLOW_PRODUCT_DELIVERY_STANDARD.md +++ b/docs/PMFLOW_PRODUCT_DELIVERY_STANDARD.md @@ -554,14 +554,15 @@ commit gate 判定： 当前最建议下一步执行： -**Phase C2：ChangeSet commit 后端与前端实现。** +**Phase C2：ChangeSet commit 前端接入。** 原因： - A3 已完成真实只读投影，Phase B 已完成真实 Project Launch 创建链。 - Phase C1 已把项目内消息、结构化候选、审查动作和证据链落到真实表。 - C2-contract 已锁定 commit API、原子提交规则、响应合同和失败语义。 -- 当前缺口是按合同把已接受/已编辑的 ChangeItem 经 commit 写入 `truth_nodes / tasks / semantic_facts`，并补齐高风险 approval gate。 +- C2-backend 已按合同实现 atomic commit route/service/schema/tests。 +- 当前缺口是前端真实模式仍未调用 `POST /pmflow-os/change-sets/{change_set_id}/commit`，需要替换 demo-only `promoteAcceptedToGraph`。 ### 7.1 Phase A 建议拆成三个小任务 @@ -611,6 +612,45 @@ commit gate 判定： ## 8. 进度台账 +### 2026-04-26 / Phase C2 Backend - ChangeSet Atomic Commit / done + +完成内容： + +- 新增 `POST /pmflow-os/change-sets/{change_set_id}/commit` 后端入口，复用项目成员/OWNER/REVIEWER 权限边界，统一响应包络。 +- 在 `PMFlowChangeReviewService` 实现 atomic commit：只处理 `accepted / edited` 项，默认提交全部 eligible items；存在未审项时阻断全量 commit；任一写入失败则回滚整个事务。 +- 将 `Requirement / Decision / Risk` 写入 `truth_nodes`，将 `Task` 写入 `truth_nodes(node_type='task') + tasks`，并为 truth node/task 写入 `semantic_facts` seed。 +- 补齐高风险 approval gate：`risk_level=high` 必须存在同 item 的 approved/edited approval 与 evidence，否则返回 `409 high_risk_approval_required` 且不写正式对象。 +- 将 `PMFlowChangeSet / PMFlowChangeItem` 序列化扩展到 `committed / partially_committed / skipped` 状态，避免 commit 后投影被误映射为 deferred。 +- 将 `backend/app/api/__init__.py` 改为轻量包入口，避免 focused API 测试导入单个模块时拉起 Feishu/MCP 等可选依赖。 + +涉及文件： + +- `backend/app/api/__init__.py` +- `backend/app/api/pmflow_os_api/__init__.py` +- `backend/app/schemas/pmflow_os.py` +- `backend/app/services/pmflow_change_review_service.py` +- `backend/tests/test_pmflow_change_review_api.py` +- `docs/architecture/PMFLOW_OS_PRODUCT_CONTRACT.md` +- `docs/architecture/CODE_REFERENCE_INDEX.md` +- `docs/PMFLOW_PRODUCT_DELIVERY_STANDARD.md` + +验证： + +- `/opt/homebrew/bin/python3.11 -m py_compile backend/app/api/__init__.py backend/app/api/pmflow_os_api/__init__.py backend/app/services/pmflow_change_review_service.py backend/app/schemas/pmflow_os.py backend/tests/test_pmflow_change_review_api.py` 通过。 +- `/opt/homebrew/bin/python3.11 -m pytest tests/test_pmflow_change_review_api.py -q` 通过：`6 passed`。 +- `/opt/homebrew/bin/python3.11 -m pytest tests/test_pmflow_change_review_api.py tests/test_pmflow_os_projection_api.py tests/test_pmflow_launch_api.py -q` 通过：`18 passed`。 + +未完成或风险： + +- 前端真实模式尚未调用 commit API，`promoteAcceptedToGraph` 在 API 模式仍不会写入项目记忆。 +- C2 V1 暂不支持 `ModelDecision / MetricSignal` commit；按合同必须 fail-closed。 +- 本地仍有未提交删除痕迹，未纳入本次提交。 +- 当前为本地开发工作区验证，未执行部署环境热重启。 + +下一步建议： + +- 执行 `C2-frontend`：在 `frontend/src/api/pmflowOs.ts` 增加 commit wrapper，并让 `frontend/src/features/pmflow-os/store.tsx` 真实模式调用 commit 后 refetch truth-state。 + ### 2026-04-26 / G0 Freeze + C2 Contract / done 完成内容： diff --git a/docs/architecture/CODE_REFERENCE_INDEX.md b/docs/architecture/CODE_REFERENCE_INDEX.md index bb253d8b..9648d39f 100644 --- a/docs/architecture/CODE_REFERENCE_INDEX.md +++ b/docs/architecture/CODE_REFERENCE_INDEX.md @@ -89,7 +89,7 @@ - SSTC 负责人推荐接口：参考 `POST /projects/{project_id}/tasks/{task_id}/recommend-assignee` - SSTC 任务更新接口：参考 `PATCH /projects/{project_id}/tasks/{task_id}` - SSTC 发布创建接口：参考 `POST /projects/{project_id}/releases` -- PMFlow OS 真实读路径与 C1 Change Review API：参考 `backend/app/api/pmflow_os_api/`、`backend/app/services/pmflow_os_projection_service.py`、`backend/app/services/pmflow_change_review_service.py`、`backend/app/schemas/pmflow_os.py`；只读提供 `GET /pmflow-os/projects/{project_id}/truth-state` 与 `GET /pmflow-os/projects/{project_id}/execution-summary`，复用 `truth_nodes / tasks / semantic_facts / ai_project_control / dev_sync`，并读取 `pmflow_threads / pmflow_thread_messages / pmflow_change_sets / pmflow_change_items / pmflow_approvals / pmflow_evidence_links`；C1 写 API 包括 thread message、`proposed_items` extraction、ChangeSet detail、ChangeItem `accept/edit/defer`，C2 commit 合同已锁定为 `POST /pmflow-os/change-sets/{change_set_id}/commit`，默认 `mode=atomic`，实现前必须先读 `docs/architecture/PMFLOW_OS_PRODUCT_CONTRACT.md` 与 `docs/architecture/API_CONTRACTS.md` +- PMFlow OS 真实读路径、C1 Change Review API 与 C2 backend commit：参考 `backend/app/api/pmflow_os_api/`、`backend/app/services/pmflow_os_projection_service.py`、`backend/app/services/pmflow_change_review_service.py`、`backend/app/schemas/pmflow_os.py`；只读提供 `GET /pmflow-os/projects/{project_id}/truth-state` 与 `GET /pmflow-os/projects/{project_id}/execution-summary`，复用 `truth_nodes / tasks / semantic_facts / ai_project_control / dev_sync`，并读取 `pmflow_threads / pmflow_thread_messages / pmflow_change_sets / pmflow_change_items / pmflow_approvals / pmflow_evidence_links`；C1 写 API 包括 thread message、`proposed_items` extraction、ChangeSet detail、ChangeItem `accept/edit/defer`；C2 backend 已实现 `POST /pmflow-os/change-sets/{change_set_id}/commit`，默认 `mode=atomic`，将 `Requirement / Decision / Risk` 写入 `truth_nodes`，将 `Task` 写入 `truth_nodes(node_type='task') + tasks`，高风险项强制 approval gate - Project Launch 真实创建链：参考 `backend/app/api/pmflow_launch_api/`、`backend/app/services/pmflow_launch_service.py`、`backend/app/schemas/pmflow_launch.py` 与模型 `backend/app/models/pmflow_launch.py`；接口为 `POST /pmflow-launch/drafts` 与 `POST /pmflow-launch/drafts/{draft_id}/confirm`，draft 只写 `pmflow_launch_drafts`，confirm 在同一事务内创建 `projects / user_projects / pmflow_threads / pmflow_thread_messages / truth_nodes / tasks / semantic_facts seed / operation_logs` - PMFlow Change Review 模型与 migration：参考 `backend/app/models/pmflow_change_review.py` 与 `backend/alembic/versions/20260426_add_pmflow_change_review_tables.py`；只包含 `pmflow_change_sets / pmflow_change_items / pmflow_approvals / pmflow_evidence_links`，本阶段不得新增或依赖 `/commit` diff --git a/docs/architecture/PMFLOW_OS_PRODUCT_CONTRACT.md b/docs/architecture/PMFLOW_OS_PRODUCT_CONTRACT.md index da1593a2..3a7006de 100644 --- a/docs/architecture/PMFLOW_OS_PRODUCT_CONTRACT.md +++ b/docs/architecture/PMFLOW_OS_PRODUCT_CONTRACT.md @@ -2,7 +2,7 @@ 版本日期：2026-04-26 -本文档是 PMFlow OS 从本地高保真主链走向真实数据闭环的产品级合同。它锁定对象真源、读写边界、状态流转和后续开发接口，避免 `frontend/src/features/pmflow-os/` 继续围绕 mock store 扩散。A3 已完成真实只读 projection，Phase B 已完成 Project Launch 创建链，Phase C1 已完成 Change Review 数据底座与审查状态机；Phase C2 commit 合同已收口，后续实现不得再自行改变 API shape。 +本文档是 PMFlow OS 从本地高保真主链走向真实数据闭环的产品级合同。它锁定对象真源、读写边界、状态流转和后续开发接口，避免 `frontend/src/features/pmflow-os/` 继续围绕 mock store 扩散。A3 已完成真实只读 projection，Phase B 已完成 Project Launch 创建链，Phase C1 已完成 Change Review 数据底座与审查状态机；Phase C2 backend 已按 atomic commit 合同实现，后续前端接入不得再自行改变 API shape。 产品方向真源仍以 `docs/PMFLOW_PRODUCT_BLUEPRINT.md` 为准；UI 行为与状态标签以 `docs/PMFLOW_UI_CONTRACT.md` 为准；推进节奏以 `docs/PMFLOW_PRODUCT_DELIVERY_STANDARD.md` 为准。 @@ -40,7 +40,7 @@ pmflow_approvals pmflow_evidence_links ``` -当前仍未实现 `/commit` 代码，因此 accepted/edited ChangeItem 不得被投影成 confirmed Project Memory；但 C2 实现必须遵守本文 4.6 和第 5 节合同。 +`/commit` 后端代码已实现；前端真实模式接入完成前，仍不得用本地 `promoteAcceptedToGraph` 冒充正式 Project Memory 写入。C2 后续前端实现必须遵守本文 4.6 和第 5 节合同。 ## 2. 对象映射表 @@ -439,8 +439,9 @@ PMFlow OS 不重造开发执行系统。执行桥复用： 1. A3 最小真实读路径：已实现 `pmflow_os_api` read-only projection，提供 `truth-state` 与 `execution-summary`。 2. Phase B：已把 `/project-launch` 的草案确认接入真实项目创建和初始 Project Memory 写入。 3. Phase C1：已实现 ChangeSet/ChangeItem/Approval/EvidenceLink 表与 `accept / edit / defer` 审查 API。 -4. Phase C2：下一步实现 ChangeSet commit，把 `accepted / edited` 写入 `truth_nodes / tasks / semantic_facts` 并补齐高风险 approval gate。 -5. Phase D：把 `ExecutionBridgePage` 切到 `execution-summary` 和 dev-sync 真实数据。 -6. Phase E：把 Release/Growth 投影接入 release validation、feedback、revenue、ROI、compliance。 +4. Phase C2 backend：已实现 ChangeSet atomic commit，把 `accepted / edited` 写入 `truth_nodes / tasks / semantic_facts` 并补齐高风险 approval gate。 +5. Phase C2 frontend：下一步把真实模式下的 `promoteAcceptedToGraph` 接到 `POST /pmflow-os/change-sets/{change_set_id}/commit`。 +6. Phase D：把 `ExecutionBridgePage` 切到 `execution-summary` 和 dev-sync 真实数据。 +7. Phase E：把 Release/Growth 投影接入 release validation、feedback、revenue、ROI、compliance。 本文档若与后续实现不一致，必须先更新合同再改代码。