From a7008bb3cc71bb500714ebfb2bac7018816b77d3 Mon Sep 17 00:00:00 2001 From: Albert Date: Sun, 26 Apr 2026 10:51:14 +0800 Subject: [PATCH] feat(pmflow-os): add execution progress writeback --- backend/app/api/pmflow_os_api/__init__.py | 34 ++++ backend/app/schemas/pmflow_os.py | 29 +++ .../services/pmflow_os_projection_service.py | 98 ++++++++- .../tests/test_pmflow_os_projection_api.py | 173 ++++++++++++++++ frontend/src/api/pmflowOs.ts | 39 ++++ .../pmflow-os/pages/ExecutionBridgePage.tsx | 186 +++++++++++++++++- .../src/features/pmflow-os/store.api.test.tsx | 60 ++++++ frontend/src/features/pmflow-os/store.tsx | 10 + 8 files changed, 620 insertions(+), 9 deletions(-) diff --git a/backend/app/api/pmflow_os_api/__init__.py b/backend/app/api/pmflow_os_api/__init__.py index 0b43db00..1c6e8de6 100644 --- a/backend/app/api/pmflow_os_api/__init__.py +++ b/backend/app/api/pmflow_os_api/__init__.py @@ -15,6 +15,7 @@ from ...schemas.pmflow_os import ( PMFlowChangeSetCommitRequest, PMFlowExecutionHandoffConsumeRequest, PMFlowExecutionHandoffRequest, + PMFlowExecutionProgressEventRequest, PMFlowExtractionCreateRequest, PMFlowThreadMessageCreateRequest, ) @@ -285,6 +286,39 @@ async def create_pmflow_os_execution_handoff( return success_response(payload.model_dump(mode="json")) +@router.post("/projects/{project_id}/execution-progress-events") +async def create_pmflow_os_execution_progress_event( + project_id: int, + body: PMFlowExecutionProgressEventRequest, + db: AsyncSession = Depends(get_db), + current_user: User = Depends(get_current_user), +): + tenant_id = int(current_user.tenant_id) + await _ensure_project_member_or_manager( + db, + tenant_id=tenant_id, + project_id=int(project_id), + current_user=current_user, + ) + service = PMFlowOSProjectionService(db, tenant_id=tenant_id) + try: + payload = await service.record_execution_progress_event( + project_id=int(project_id), + body=body, + current_user_id=int(current_user.id), + ) + except ValueError as exc: + code = str(exc) + if code == "task_not_found": + return JSONResponse(status_code=404, content=error_response(404, "task_not_found")) + if code == "repo_binding_not_found": + return JSONResponse(status_code=422, content=error_response(422, "repo_binding_not_found")) + if code == "project_not_found": + raise HTTPException(status_code=404, detail="project_not_found") from exc + raise + return success_response(payload.model_dump(mode="json")) + + @router.post("/projects/{project_id}/execution-handoffs/{prompt_package_id}/consume") async def consume_pmflow_os_execution_handoff( project_id: int, diff --git a/backend/app/schemas/pmflow_os.py b/backend/app/schemas/pmflow_os.py index eedcedfd..1a74475f 100644 --- a/backend/app/schemas/pmflow_os.py +++ b/backend/app/schemas/pmflow_os.py @@ -13,6 +13,20 @@ PMFlowRiskLevel = Literal["low", "medium", "high"] PMFlowExecutionStatus = Literal["queued", "pushed", "coding", "pr_open", "testing", "synced", "failed"] PMFlowReleaseStatus = Literal["check_pending", "ready", "shipping", "live", "watch"] PMFlowExecutionRuntimeTarget = Literal["local_bridge", "codex", "github_agent", "manual"] +PMFlowExecutionProgressSource = Literal["local_bridge", "codex", "github_agent", "manual"] +PMFlowExecutionProgressEventType = Literal[ + "branch_created", + "commit", + "pr_opened", + "pr_synced", + "test_passed", + "test_failed", + "checks_passed", + "blocked", + "merged", + "accepted", + "manual_progress", +] 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", "committed", "skipped"] @@ -100,6 +114,21 @@ class PMFlowExecutionHandoffConsumeRequest(BaseModel): note: Optional[str] = Field(default=None, max_length=4000) +class PMFlowExecutionProgressEventRequest(BaseModel): + task_id: str = Field(min_length=1) + event_type: PMFlowExecutionProgressEventType + source: PMFlowExecutionProgressSource = "manual" + repo_binding_id: Optional[str] = None + branch_name: Optional[str] = Field(default=None, max_length=255) + commit_sha: Optional[str] = Field(default=None, max_length=255) + pr_number: Optional[int] = Field(default=None, ge=1) + test_summary: Optional[str] = Field(default=None, max_length=4000) + workflow_state: Optional[str] = Field(default=None, max_length=120) + summary: Optional[str] = Field(default=None, max_length=4000) + event_source_id: Optional[str] = Field(default=None, max_length=255) + occurred_at: 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_os_projection_service.py b/backend/app/services/pmflow_os_projection_service.py index e658c13b..5a0c988e 100644 --- a/backend/app/services/pmflow_os_projection_service.py +++ b/backend/app/services/pmflow_os_projection_service.py @@ -39,6 +39,7 @@ from ..schemas.pmflow_os import ( PMFlowDecision, PMFlowExecutionHandoffConsumeRequest, PMFlowExecutionHandoffRequest, + PMFlowExecutionProgressEventRequest, PMFlowExecutionContext, PMFlowFeedEvent, PMFlowMetricSignal, @@ -322,6 +323,33 @@ class PMFlowOSProjectionService: await self.db.commit() return await self.get_execution_summary(project_id=project_id, current_user_id=current_user_id) + async def record_execution_progress_event( + self, + *, + project_id: int, + body: PMFlowExecutionProgressEventRequest, + current_user_id: Optional[int] = None, + ) -> ExecutionSummaryProjection: + await self._load_project(project_id=project_id) + task = await self._load_project_task(project_id=project_id, task_id=body.task_id) + repo = await self._load_project_repo_binding(project_id=project_id, repo_binding_id=body.repo_binding_id) + payload = self._progress_event_payload(body=body, repo=repo) + summary = self._progress_event_summary(body=body, task=task, repo=repo) + control = AIProjectControlService(self.db, tenant_id=self.tenant_id) + await control.create_or_apply_progress_event( + task_id=int(task.id), + project_id=int(project_id), + event_type=body.event_type, + source=body.source, + payload=payload, + summary=summary, + user_id=int(current_user_id) if current_user_id is not None else None, + repo_binding_id=int(repo.id) if repo is not None else None, + event_source_id=body.event_source_id, + occurred_at=body.occurred_at, + ) + return await self.get_execution_summary(project_id=project_id, current_user_id=current_user_id) + async def _load_project(self, *, project_id: int) -> Project: result = await self.db.execute( select(Project).where(and_(Project.id == int(project_id), Project.tenant_id == self.tenant_id)) @@ -1108,18 +1136,18 @@ class PMFlowOSProjectionService: workflow_state = str((event or {}).get("workflow_state") or "").lower() derived = str((event or {}).get("derived_task_status") or task_status or "").lower() combined = " ".join([event_type, workflow_state, derived]) + if "failed" in combined or "blocked" in combined: + return "failed" + if "merged" in combined or "accepted" in combined or "done" in combined or "completed" in combined or "synced" in combined: + return "synced" if "test" in combined: return "testing" if "pull_request" in combined or "pr" in combined or "review" in combined: return "pr_open" if "commit" in combined or "coding" in combined or "in_progress" in combined: return "coding" - if "pushed" in combined or "queued" in combined: + if "branch" in combined or "pushed" in combined or "queued" in combined: return "pushed" - if "done" in combined or "completed" in combined or "synced" in combined: - return "synced" - if "failed" in combined or "blocked" in combined: - return "failed" return "queued" def _pick_execution_task_id( @@ -1173,7 +1201,9 @@ class PMFlowOSProjectionService: def _project_test_summary(self, events: list[dict[str, Any]]) -> str: for event in events: - if "test" in str(event.get("event_type") or "").lower(): + event_type = str(event.get("event_type") or "").lower() + workflow_state = str(event.get("workflow_state") or "").lower() + if "test" in event_type or "check" in event_type or workflow_state == "testing": return str(event.get("summary") or event.get("workflow_state") or "test event recorded") return "pending" @@ -1370,6 +1400,62 @@ class PMFlowOSProjectionService: def _handoff_consumed_summary(self, *, task: Task, runtime_target: str, consumer_label: str) -> str: return f"执行交接已领取：{task.title} -> {runtime_target} / {consumer_label}" + def _progress_event_payload( + self, + *, + body: PMFlowExecutionProgressEventRequest, + repo: Optional[RepoBinding], + ) -> dict[str, Any]: + payload: dict[str, Any] = { + "workflow_state": body.workflow_state or self._default_workflow_state(body.event_type), + "source": body.source, + "boundary": "PMFlow records execution result metadata only; command execution stays in Local Bridge, Codex, GitHub agent, or manual runtime.", + } + for key, value in { + "branch_name": body.branch_name, + "commit_sha": body.commit_sha, + "pr_number": body.pr_number, + "test_summary": body.test_summary, + "event_source_id": body.event_source_id, + }.items(): + if value is not None and str(value).strip(): + payload[key] = value + if repo is not None: + payload["repo_binding_id"] = str(repo.id) + payload["repository_full_name"] = f"{repo.owner}/{repo.repo_name}" + return payload + + def _progress_event_summary( + self, + *, + body: PMFlowExecutionProgressEventRequest, + task: Task, + repo: Optional[RepoBinding], + ) -> str: + if body.summary: + return body.summary + if body.test_summary: + return body.test_summary + repo_label = f"{repo.owner}/{repo.repo_name}" if repo is not None else "未绑定仓库" + detail = body.commit_sha or (f"PR #{body.pr_number}" if body.pr_number else body.branch_name) or self._default_workflow_state(body.event_type) + return f"执行结果回写：{task.title} / {body.event_type} / {repo_label} / {detail}" + + def _default_workflow_state(self, event_type: str) -> str: + value = str(event_type or "").strip().lower() + if value == "branch_created": + return "pushed" + if value == "commit": + return "coding" + if value in {"pr_opened", "pr_synced"}: + return "in_review" + if value in {"test_passed", "test_failed", "checks_passed"}: + return "testing" + if value in {"merged", "accepted"}: + return "synced" + if value == "blocked": + return "failed" + return "manual_progress" + def _latest_iso(self, values: list[Optional[datetime]]) -> Optional[str]: valid = [value for value in values if value is not None] if not valid: diff --git a/backend/tests/test_pmflow_os_projection_api.py b/backend/tests/test_pmflow_os_projection_api.py index 311c49c9..4165709f 100644 --- a/backend/tests/test_pmflow_os_projection_api.py +++ b/backend/tests/test_pmflow_os_projection_api.py @@ -599,6 +599,179 @@ async def test_execution_handoff_consume_records_pickup_event(db_env) -> None: assert event_types == ["handoff_created", "handoff_consumed"] +@pytest.mark.asyncio +async def test_execution_progress_writeback_records_branch_pr_commit_and_test(db_env) -> None: + session = db_env + user, project = await _seed_user_project(session) + charter = TruthNode( + tenant_id=1, + node_type="charter", + project_id=project.id, + created_by=user.id, + title="执行回写宪章", + content={"business_goal": "把运行端结果回流到执行摘要"}, + version=1, + status="aligned", + ) + session.add(charter) + await session.flush() + task = Task( + tenant_id=1, + truth_node_id=charter.id, + project_id=project.id, + assignee_id=user.id, + ai_recommended_assignees=None, + ai_breakdown_payload={"acceptance_criteria": ["回流 branch/commit/PR/test"]}, + manual_overrides=None, + title="执行结果写回", + description="Codex 或 Local Bridge 写回进展", + required_skill_tag_ids=[], + status="confirmed", + priority="high", + ) + repo = RepoBinding( + tenant_id=1, + project_id=project.id, + provider="github", + owner="pmflow", + repo_name="writeback", + default_branch="main", + installation_id="install-4", + status="active", + ) + session.add_all([task, repo]) + await session.commit() + 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: + commit_response = await client.post( + f"/pmflow-os/projects/{project.id}/execution-progress-events", + json={ + "task_id": str(task.id), + "repo_binding_id": str(repo.id), + "event_type": "commit", + "source": "codex", + "branch_name": "codex/main-flow-d3", + "commit_sha": "abc123def456", + "workflow_state": "coding", + }, + ) + pr_response = await client.post( + f"/pmflow-os/projects/{project.id}/execution-progress-events", + json={ + "task_id": str(task.id), + "repo_binding_id": str(repo.id), + "event_type": "pr_opened", + "source": "github_agent", + "branch_name": "codex/main-flow-d3", + "pr_number": 77, + }, + ) + test_response = await client.post( + f"/pmflow-os/projects/{project.id}/execution-progress-events", + json={ + "task_id": str(task.id), + "repo_binding_id": str(repo.id), + "event_type": "test_failed", + "source": "local_bridge", + "branch_name": "codex/main-flow-d3", + "test_summary": "单测失败：ExecutionBridgePage 回写表单缺少断言", + }, + ) + + assert commit_response.status_code == 200 + assert commit_response.json()["data"]["execution"]["status"] == "coding" + assert pr_response.status_code == 200 + pr_payload = pr_response.json()["data"] + assert pr_payload["execution"]["status"] == "pr_open" + assert pr_payload["execution"]["prStatus"] == "PR #77 · linked" + assert test_response.status_code == 200 + test_payload = test_response.json()["data"] + assert test_payload["execution"]["status"] == "failed" + assert test_payload["execution"]["testSummary"] == "单测失败：ExecutionBridgePage 回写表单缺少断言" + assert test_payload["recentProgressEvents"][0]["event_type"] == "test_failed" + assert test_payload["recentProgressEvents"][0]["repo"]["full_name"] == "pmflow/writeback" + + link = (await session.execute(select(TaskCodeLink).where(TaskCodeLink.task_id == task.id))).scalar_one() + assert link.branch_name == "codex/main-flow-d3" + assert link.pr_number == 77 + assert link.last_commit_sha == "abc123def456" + events = (await session.execute(select(ProgressEvent).where(ProgressEvent.task_id == task.id))).scalars().all() + assert [event.event_type for event in events] == ["commit", "pr_opened", "test_failed"] + + +@pytest.mark.asyncio +async def test_execution_progress_writeback_rejects_repo_outside_project(db_env) -> None: + session = db_env + user, project = await _seed_user_project(session) + other_project = Project( + tenant_id=1, + name="其它回写项目", + title="其它回写项目", + description="不能被当前项目进展事件引用", + owner_id=user.id, + status="active", + assigned_role="BUILDER", + signal_count=0, + ) + session.add(other_project) + await session.flush() + session.add(UserProject(tenant_id=1, user_id=user.id, project_id=other_project.id)) + charter = TruthNode( + tenant_id=1, + node_type="charter", + project_id=project.id, + created_by=user.id, + title="当前项目宪章", + content={"business_goal": "校验 repo project 边界"}, + version=1, + status="aligned", + ) + repo = RepoBinding( + tenant_id=1, + project_id=other_project.id, + provider="github", + owner="pmflow", + repo_name="other-writeback", + default_branch="main", + status="active", + ) + session.add_all([charter, repo]) + await session.flush() + task = Task( + tenant_id=1, + truth_node_id=charter.id, + project_id=project.id, + assignee_id=user.id, + ai_recommended_assignees=None, + ai_breakdown_payload=None, + manual_overrides=None, + title="当前项目任务", + description="不能绑定其它项目 repo", + required_skill_tag_ids=[], + status="confirmed", + priority="medium", + ) + session.add(task) + await session.commit() + 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: + response = await client.post( + f"/pmflow-os/projects/{project.id}/execution-progress-events", + json={ + "task_id": str(task.id), + "repo_binding_id": str(repo.id), + "event_type": "commit", + "source": "codex", + "branch_name": "codex/wrong-repo", + }, + ) + + assert response.status_code == 422 + assert response.json()["message"] == "repo_binding_not_found" + + @pytest.mark.asyncio async def test_execution_handoff_rejects_task_outside_project(db_env) -> None: session = db_env diff --git a/frontend/src/api/pmflowOs.ts b/frontend/src/api/pmflowOs.ts index 8b3839ec..a3933394 100644 --- a/frontend/src/api/pmflowOs.ts +++ b/frontend/src/api/pmflowOs.ts @@ -23,6 +23,19 @@ export interface PMFlowExecutionSummary { } export type PMFlowExecutionRuntimeTarget = "local_bridge" | "codex" | "github_agent" | "manual"; +export type PMFlowExecutionProgressSource = "local_bridge" | "codex" | "github_agent" | "manual"; +export type PMFlowExecutionProgressEventType = + | "branch_created" + | "commit" + | "pr_opened" + | "pr_synced" + | "test_passed" + | "test_failed" + | "checks_passed" + | "blocked" + | "merged" + | "accepted" + | "manual_progress"; export interface PMFlowExecutionHandoffRequest { task_id: string; @@ -37,6 +50,21 @@ export interface PMFlowExecutionHandoffConsumeRequest { note?: string | null; } +export interface PMFlowExecutionProgressEventRequest { + task_id: string; + event_type: PMFlowExecutionProgressEventType; + source?: PMFlowExecutionProgressSource; + repo_binding_id?: string | null; + branch_name?: string | null; + commit_sha?: string | null; + pr_number?: number | null; + test_summary?: string | null; + workflow_state?: string | null; + summary?: string | null; + event_source_id?: string | null; + occurred_at?: string | null; +} + export interface PMFlowPromptPackage { id: string; project_id: string; @@ -195,6 +223,17 @@ export async function consumePMFlowExecutionHandoff( return unwrap(data); } +export async function createPMFlowExecutionProgressEvent( + projectId: string, + payload: PMFlowExecutionProgressEventRequest +): Promise { + const { data } = await api.post>( + `/pmflow-os/projects/${projectId}/execution-progress-events`, + payload + ); + return unwrap(data); +} + export async function createPMFlowThreadMessage( projectId: string, threadId: string, diff --git a/frontend/src/features/pmflow-os/pages/ExecutionBridgePage.tsx b/frontend/src/features/pmflow-os/pages/ExecutionBridgePage.tsx index e6cc2045..e20951b2 100644 --- a/frontend/src/features/pmflow-os/pages/ExecutionBridgePage.tsx +++ b/frontend/src/features/pmflow-os/pages/ExecutionBridgePage.tsx @@ -1,3 +1,4 @@ +import { useState } from "react"; import { AlertTriangle, CheckCircle2, @@ -17,6 +18,8 @@ import { import type { PMFlowDevTask, + PMFlowExecutionProgressEventType, + PMFlowExecutionProgressSource, PMFlowExecutionSummary, PMFlowExecutionRuntimeTarget, PMFlowProgressEvent, @@ -54,6 +57,7 @@ export default function ExecutionBridgePage() { refetchExecutionSummary, createExecutionHandoff, consumeExecutionHandoff, + recordExecutionProgress, advanceExecution, } = usePMFlowOsStore(); const demoSummary = dataSourceMode === "demo" ? createDemoExecutionSummary(state) : null; @@ -111,7 +115,10 @@ export default function ExecutionBridgePage() { onConsumeHandoff={isDemo ? undefined : consumeExecutionHandoff} onDemoPush={isDemo ? advanceExecution : undefined} /> - + @@ -313,7 +320,24 @@ function RepoBindingList({ repos }: { repos: PMFlowRepoBinding[] }) { ); } -function ExecutionReviewPanel({ summary }: { summary: PMFlowExecutionSummary }) { +function ExecutionReviewPanel({ + summary, + onRecordProgress, +}: { + summary: PMFlowExecutionSummary; + onRecordProgress?: (payload: { + task_id: string; + event_type: PMFlowExecutionProgressEventType; + source?: PMFlowExecutionProgressSource; + repo_binding_id?: string | null; + branch_name?: string | null; + commit_sha?: string | null; + pr_number?: number | null; + test_summary?: string | null; + workflow_state?: string | null; + summary?: string | null; + }) => void; +}) { return (
@@ -326,7 +350,10 @@ function ExecutionReviewPanel({ summary }: { summary: PMFlowExecutionSummary })
- +
+ + {onRecordProgress ? : null} +
); @@ -388,6 +415,159 @@ function ExecutionFacts({ summary }: { summary: PMFlowExecutionSummary }) { ); } +function ExecutionWritebackPanel({ + summary, + onRecordProgress, +}: { + summary: PMFlowExecutionSummary; + onRecordProgress: (payload: { + task_id: string; + event_type: PMFlowExecutionProgressEventType; + source?: PMFlowExecutionProgressSource; + repo_binding_id?: string | null; + branch_name?: string | null; + commit_sha?: string | null; + pr_number?: number | null; + test_summary?: string | null; + workflow_state?: string | null; + summary?: string | null; + }) => void; +}) { + const task = summary.activeDevTasks.find((item) => item.id === summary.execution.taskId) || summary.activeDevTasks[0] || null; + const repo = task?.code_link?.repo || summary.repoBindings[0] || null; + const [eventType, setEventType] = useState("commit"); + const [source, setSource] = useState("codex"); + const [branchName, setBranchName] = useState(task?.code_link?.branch_name || ""); + const [commitSha, setCommitSha] = useState(task?.code_link?.last_commit_sha || ""); + const [prNumber, setPrNumber] = useState(task?.code_link?.pr_number ? String(task.code_link.pr_number) : ""); + const [testSummary, setTestSummary] = useState(""); + const [workflowState, setWorkflowState] = useState(""); + const canSubmit = Boolean(task); + + return ( +
+
+ +
+

执行结果回写

+

+ 记录 Codex / Local Bridge / GitHub agent 已产生的结果；这里只写 progress event，不执行命令。 +

+ {!task ? ( +
缺少 activeDevTasks，无法写回执行结果。
+ ) : null} + {task && !repo ? ( +
当前没有 repoBinding，将只记录进展事件，不绑定 branch/PR/commit 到仓库。
+ ) : null} +
+ + +
+ + +
+
+ + +
+