#!/usr/bin/env python3 """Generate the P1R API coverage matrix. The script is intentionally conservative: if it cannot prove an operation has a dedicated business implementation, it marks the operation as incomplete or blocked. This prevents catch-all contract handlers from being counted as done. """ from __future__ import annotations import argparse import json import hashlib import re from dataclasses import asdict, dataclass from pathlib import Path from typing import Any try: import yaml except ImportError as exc: raise SystemExit("PyYAML is required. Install it with: python3 -m pip install PyYAML") from exc REPO_ROOT = Path(__file__).resolve().parents[2] CONTRACT_ROOT = REPO_ROOT / "docs" / "api-contracts" MUSE_CLOUD_ROOT = REPO_ROOT / "muse-cloud" REPORT_ROOT = REPO_ROOT / "docs" / "superpowers" / "reports" JSON_REPORT = REPORT_ROOT / "p1r-api-coverage.json" MARKDOWN_REPORT = REPORT_ROOT / "p1r-api-coverage.md" HTTP_METHODS = {"get", "post", "put", "patch", "delete"} NON_REAL_IMPLEMENTATION_STATUSES = {"catch_all", "generic_persistence", "sse_placeholder", "missing"} NON_REAL_COMPLETION_STATUSES = {"incomplete", "blocked"} JAVA_HTTP_METHODS = {method.upper() for method in HTTP_METHODS} MAPPING_METHODS = { "Get": "GET", "Post": "POST", "Put": "PUT", "Patch": "PATCH", "Delete": "DELETE", } SSE_PLACEHOLDER_READY_MARKERS = { "muse ai task stream ready", "muse event stream ready", } APPROVED_COMPLETED_DOMAINS = {"ai", "knowledge"} APPROVED_COMPLETED_OPERATIONS_BY_DOMAIN = { # Account / Content / Market 在 2026-06-17 经真实 PG 证据 + 用户批准后仍保持 operation-level approval。 "account": { "adminCreateCallAttributionJob", "adminCreateNewApiBinding", "adminCreateQuotaAdjustment", "adminCreateQuotaRequest", "adminGetBalanceSnapshots", "adminGetCallAttributionJob", "adminGetIntegrationCallByCorrelation", "adminGetUserEntitlements", "adminListAccountUsers", "adminListNewApiBindings", "adminListPurchaseRecords", "adminListQuotaAdjustments", "adminListUsageRecords", "appAcknowledgeSecurityEvent", "appCreateExportTask", "appCreateQuotaRequest", "appDownloadExport", "appGetExportTask", "appGetIntegrationCallByCorrelation", "appGetQuotaRequest", "appGetSecurityEvent", "appListLicenses", "appListPublishRecords", "appListPurchases", "appListSecurityEvents", "appRecheckNewApiBinding", "getAppBalanceSnapshots", "getAppEntitlements", "getAppNewApiBinding", "getAppUsage", "getCurrentUser", "getProfile", "updateProfile", }, "content": { "adminGetWork", "adminListChapters", "adminListExportTasks", "adminListImportTasks", "adminListWorks", "adminRiskAction", "batchConfirmChapters", "confirmChapterParseResult", "confirmPlanningCandidate", "createBlock", "createChapter", "createExportTask", "createImportTask", "createParseJob", "createPlanningCandidate", "createStyleCheck", "createWork", "deleteBlock", "deleteChapter", "deleteWork", "discardPlanningCandidate", "downloadExportPackage", "exportWork", "getBlock", "getBlockSourceAttribution", "listBlockRevisions", "getChapter", "getExportTask", "getImportTask", "getMetaProjection", "getParseJob", "getPlanning", "getPlanningCandidate", "getStyleCheckResult", "getWork", "listBlocks", "listChapters", "listMetaProjections", "listParseJobChapters", "listPlanningCandidates", "listWorks", "mergeBlockSuggestion", "mergeBlocks", "rejectChapterParseResult", "reorderChapters", "retryParseJob", "saveBlock", "savePlanningItem", "splitBlock", "updateChapter", "updateWork", "validateDynamicFields", }, "events": {"streamEvents"}, "market": { "adminApprovePublishRequest", "adminDelistAsset", "adminGetAppeal", "adminGetMarketAsset", "adminListAppeals", "adminListMarketAssets", "adminListPublishRequests", "adminPreviewGovernanceImpact", "adminRecallAsset", "adminRejectPublishRequest", "adminResolveAppeal", "cancelHandoff", "createBindPrecheck", "createMarketplaceHandoff", "favoriteAsset", "getGovernanceImpact", "getHandoffStatus", "getMarketplaceAsset", "installMarketplaceAsset", "listMarketplaceAssets", "listMarketplaceCategories", "listMarketplaceRecommendations", "listMyPublishRecords", "purchaseAsset", "runPublishCheck", "savePublishDraft", "submitAppeal", "submitPublishRequest", "supplementAppeal", "unfavoriteAsset", "withdrawAppeal", "withdrawPublishRequest", }, "meta": { "activateFunctionChainVersion", "activateMetaSchemaVersion", "deprecateMetaSchemaVersion", "getMetaSchema", "getMetaSchemaVersion", "getProtectionNode", "listFunctionChains", "listMetaSchemas", "listProtectionNodes", "previewFunctionChainImpact", "previewMetaSchemaDraftImpact", "publishMetaSchemaDraft", "rollbackMetaSchemaVersion", "saveMetaSchemaDraft", "setMetaSchemaGrayRules", "validateMetaSchemaDraft", }, } APPROVED_COMPLETED_OPERATIONS = { f"{domain}:{operation_id}" for domain, operation_ids in APPROVED_COMPLETED_OPERATIONS_BY_DOMAIN.items() for operation_id in operation_ids } AI_DOMAIN_TEST_FILES = ( "muse-cloud/muse-server/src/test/java/cn/iocoder/muse/server/framework/api/P1rAiRuntimeEndToEndLiveAcceptanceIT.java", "muse-cloud/muse-module-ai/muse-module-ai-server/src/test/java/cn/iocoder/muse/module/ai/application/muse/facade/P1rNewApiLiveAcceptanceIT.java", ) # S6d(commit 302603d9)删除 RAGFlow 运行时并以 Dify 转默认装配,原 RAGFlow/runtime live 验收 IT # (P1rKnowledgeRuntimeEndToEndLiveAcceptanceIT / P1rRagFlowLiveAcceptanceIT)随之删除。 # 覆盖证据改指向其现存继任:Dify 运行时 live 契约 IT(RAGFlow live 的同目录后继)+ knowledge 路由归属测试。 KNOWLEDGE_DOMAIN_TEST_FILES = ( "muse-cloud/muse-module-knowledge/muse-module-knowledge-server/src/test/java/cn/iocoder/muse/module/knowledge/application/muse/facade/P1rDifyDatasetsContractLiveIT.java", "muse-cloud/muse-server/src/test/java/cn/iocoder/muse/server/framework/api/P1rKnowledgeRouteOwnershipTest.java", ) TEST_FILES_BY_OPERATION = { "events:streamEvents": ( "muse-cloud/muse-module-events/muse-module-events-server/src/test/java/cn/iocoder/muse/module/events/application/stream/EventsStreamServiceTest.java", ), "meta:listMetaSchemas": ( "muse-cloud/muse-module-meta/muse-module-meta-server/src/test/java/cn/iocoder/muse/module/meta/controller/admin/AdminMetaSchemaControllerTest.java", "muse-cloud/muse-module-meta/muse-module-meta-server/src/test/java/cn/iocoder/muse/module/meta/application/MetaSchemaServiceTest.java", ), "meta:getMetaSchema": ( "muse-cloud/muse-module-meta/muse-module-meta-server/src/test/java/cn/iocoder/muse/module/meta/controller/admin/AdminMetaSchemaControllerTest.java", "muse-cloud/muse-module-meta/muse-module-meta-server/src/test/java/cn/iocoder/muse/module/meta/application/MetaSchemaServiceTest.java", ), "meta:getMetaSchemaVersion": ( "muse-cloud/muse-module-meta/muse-module-meta-server/src/test/java/cn/iocoder/muse/module/meta/controller/admin/AdminMetaSchemaControllerTest.java", "muse-cloud/muse-module-meta/muse-module-meta-server/src/test/java/cn/iocoder/muse/module/meta/application/MetaSchemaServiceTest.java", ), "meta:saveMetaSchemaDraft": ( "muse-cloud/muse-module-meta/muse-module-meta-server/src/test/java/cn/iocoder/muse/module/meta/controller/admin/AdminMetaSchemaControllerTest.java", "muse-cloud/muse-module-meta/muse-module-meta-server/src/test/java/cn/iocoder/muse/module/meta/application/MetaSchemaServiceTest.java", ), "meta:validateMetaSchemaDraft": ( "muse-cloud/muse-module-meta/muse-module-meta-server/src/test/java/cn/iocoder/muse/module/meta/controller/admin/AdminMetaSchemaControllerTest.java", "muse-cloud/muse-module-meta/muse-module-meta-server/src/test/java/cn/iocoder/muse/module/meta/application/MetaSchemaValidationServiceTest.java", ), "meta:previewMetaSchemaDraftImpact": ( "muse-cloud/muse-module-meta/muse-module-meta-server/src/test/java/cn/iocoder/muse/module/meta/controller/admin/AdminMetaSchemaControllerTest.java", "muse-cloud/muse-module-meta/muse-module-meta-server/src/test/java/cn/iocoder/muse/module/meta/application/MetaSchemaImpactPreviewServiceTest.java", ), "meta:publishMetaSchemaDraft": ( "muse-cloud/muse-module-meta/muse-module-meta-server/src/test/java/cn/iocoder/muse/module/meta/controller/admin/AdminMetaSchemaControllerTest.java", "muse-cloud/muse-module-meta/muse-module-meta-server/src/test/java/cn/iocoder/muse/module/meta/application/MetaSchemaServiceTest.java", ), "meta:activateMetaSchemaVersion": ( "muse-cloud/muse-module-meta/muse-module-meta-server/src/test/java/cn/iocoder/muse/module/meta/controller/admin/AdminMetaSchemaControllerTest.java", "muse-cloud/muse-module-meta/muse-module-meta-server/src/test/java/cn/iocoder/muse/module/meta/application/MetaSchemaServiceTest.java", ), "meta:rollbackMetaSchemaVersion": ( "muse-cloud/muse-module-meta/muse-module-meta-server/src/test/java/cn/iocoder/muse/module/meta/controller/admin/AdminMetaSchemaControllerTest.java", "muse-cloud/muse-module-meta/muse-module-meta-server/src/test/java/cn/iocoder/muse/module/meta/application/MetaSchemaServiceTest.java", ), "meta:deprecateMetaSchemaVersion": ( "muse-cloud/muse-module-meta/muse-module-meta-server/src/test/java/cn/iocoder/muse/module/meta/controller/admin/AdminMetaSchemaControllerTest.java", "muse-cloud/muse-module-meta/muse-module-meta-server/src/test/java/cn/iocoder/muse/module/meta/application/MetaSchemaServiceTest.java", ), "meta:setMetaSchemaGrayRules": ( "muse-cloud/muse-module-meta/muse-module-meta-server/src/test/java/cn/iocoder/muse/module/meta/controller/admin/AdminMetaSchemaControllerTest.java", "muse-cloud/muse-module-meta/muse-module-meta-server/src/test/java/cn/iocoder/muse/module/meta/application/MetaSchemaServiceTest.java", ), "meta:listProtectionNodes": ( "muse-cloud/muse-module-meta/muse-module-meta-server/src/test/java/cn/iocoder/muse/module/meta/controller/admin/AdminProtectionNodeControllerTest.java", "muse-cloud/muse-module-meta/muse-module-meta-server/src/test/java/cn/iocoder/muse/module/meta/application/ProtectionNodeServiceTest.java", ), "meta:getProtectionNode": ( "muse-cloud/muse-module-meta/muse-module-meta-server/src/test/java/cn/iocoder/muse/module/meta/controller/admin/AdminProtectionNodeControllerTest.java", "muse-cloud/muse-module-meta/muse-module-meta-server/src/test/java/cn/iocoder/muse/module/meta/application/ProtectionNodeServiceTest.java", ), "meta:listFunctionChains": ( "muse-cloud/muse-module-meta/muse-module-meta-server/src/test/java/cn/iocoder/muse/module/meta/controller/admin/AdminFunctionChainControllerTest.java", "muse-cloud/muse-module-meta/muse-module-meta-server/src/test/java/cn/iocoder/muse/module/meta/application/FunctionChainServiceTest.java", ), "meta:previewFunctionChainImpact": ( "muse-cloud/muse-module-meta/muse-module-meta-server/src/test/java/cn/iocoder/muse/module/meta/controller/admin/AdminFunctionChainControllerTest.java", "muse-cloud/muse-module-meta/muse-module-meta-server/src/test/java/cn/iocoder/muse/module/meta/application/FunctionChainServiceTest.java", ), "meta:activateFunctionChainVersion": ( "muse-cloud/muse-module-meta/muse-module-meta-server/src/test/java/cn/iocoder/muse/module/meta/controller/admin/AdminFunctionChainControllerTest.java", "muse-cloud/muse-module-meta/muse-module-meta-server/src/test/java/cn/iocoder/muse/module/meta/application/FunctionChainServiceTest.java", ), } @dataclass(frozen=True) class Operation: domain: str side: str method: str path: str operationId: str isWrite: bool requiresCommandId: bool externalDependencies: list[str] targetStage: str implementationStatus: str completionStatus: str controllerFiles: list[str] serviceFiles: list[str] testFiles: list[str] notes: str @dataclass(frozen=True) class RouteMapping: methods: tuple[str, ...] routes: tuple[str, ...] def main() -> int: parser = argparse.ArgumentParser(description="Generate P1R API coverage matrix") parser.add_argument("--check", action="store_true", help="validate generated report after writing it") args = parser.parse_args() operations = collect_operations() REPORT_ROOT.mkdir(parents=True, exist_ok=True) report = build_report(operations) JSON_REPORT.write_text(json.dumps(report, ensure_ascii=False, indent=2) + "\n", encoding="utf-8") MARKDOWN_REPORT.write_text(render_markdown(report), encoding="utf-8") if args.check: validate_report(report) print(f"Generated {JSON_REPORT.relative_to(REPO_ROOT)}") print(f"Generated {MARKDOWN_REPORT.relative_to(REPO_ROOT)}") return 0 def collect_operations() -> list[Operation]: source_index = build_source_index() operations: list[Operation] = [] for contract_path in sorted(CONTRACT_ROOT.glob("*/openapi.yaml")): domain = contract_path.parent.name with contract_path.open("r", encoding="utf-8") as handle: document = yaml.safe_load(handle) for path, path_item in sorted((document.get("paths") or {}).items()): if not isinstance(path_item, dict): continue for method, operation_doc in sorted(path_item.items()): if method.lower() not in HTTP_METHODS or not isinstance(operation_doc, dict): continue operation_id = operation_doc.get("operationId") if not operation_id: raise SystemExit(f"Missing operationId in {contract_path}:{path}:{method}") full_path = normalize_path(path) side = detect_side(full_path) is_write = method.upper() not in {"GET", "HEAD", "OPTIONS"} requires_command_id = operation_requires_command_id(operation_doc, document) implementation = classify_implementation(full_path, side, method.upper(), operation_id, domain, source_index) operations.append( Operation( domain=domain, side=side, method=method.upper(), path=full_path, operationId=operation_id, isWrite=is_write, requiresCommandId=requires_command_id, externalDependencies=detect_external_dependencies(domain, full_path, operation_id), targetStage=target_stage(domain), implementationStatus=implementation["implementationStatus"], completionStatus=implementation["completionStatus"], controllerFiles=implementation["controllerFiles"], serviceFiles=implementation["serviceFiles"], testFiles=evidence_test_files_for(domain, operation_id, implementation["completionStatus"]), notes=implementation["notes"], ) ) return operations def build_source_index() -> dict[str, Any]: java_files = sorted(MUSE_CLOUD_ROOT.glob("**/src/main/java/**/*.java")) contract_support_files: list[str] = [] generic_persistence_files: list[str] = [] sse_placeholder_mappings: dict[str, list[RouteMapping]] = {} dedicated_controllers: list[str] = [] service_files: list[str] = [] all_text: dict[str, str] = {} for java_file in java_files: text = java_file.read_text(encoding="utf-8", errors="ignore") rel = str(java_file.relative_to(REPO_ROOT)) all_text[rel] = text if "MuseApiContractSupport.handle" in text: contract_support_files.append(rel) if "MuseContractPersistenceService" in text or "contractPersistenceService.handle" in text: generic_persistence_files.append(rel) placeholder_mappings = sse_placeholder_route_mappings(text) if placeholder_mappings: sse_placeholder_mappings[rel] = placeholder_mappings if "@RestController" in text and "MuseApiContractSupport.handle" not in text and "contractPersistenceService.handle" not in text: dedicated_controllers.append(rel) if "/application/" in rel or "/service/" in rel: service_files.append(rel) return { "allText": all_text, "contractSupportFiles": contract_support_files, "genericPersistenceFiles": generic_persistence_files, "ssePlaceholderMappings": sse_placeholder_mappings, "dedicatedControllers": dedicated_controllers, "serviceFiles": service_files, } def classify_implementation(path: str, side: str, method: str, operation_id: str, domain: str, source_index: dict[str, Any]) -> dict[str, Any]: matching_dedicated = files_containing_route_or_operation( source_index["dedicatedControllers"], source_index["allText"], path, side, method, operation_id, allow_operation_id_fallback=False, ) matching_services = files_matching_domain_services(source_index["serviceFiles"], domain) matching_catch_all = files_containing_route_or_operation( source_index["contractSupportFiles"], source_index["allText"], path, side, method, operation_id, allow_operation_id_fallback=False, ) matching_generic = files_containing_route_or_operation( source_index["genericPersistenceFiles"], source_index["allText"], path, side, method, operation_id, allow_operation_id_fallback=True, ) matching_sse = files_containing_sse_placeholder_route(source_index["ssePlaceholderMappings"], path, side, method) if matching_sse: return status( "sse_placeholder", "incomplete", matching_sse, matching_services, "SSE endpoint returns placeholder events and needs P1R real task stream implementation.", ) if matching_dedicated: dedicated_completion, dedicated_notes = dedicated_status_for(domain, operation_id) return status( "dedicated", dedicated_completion, matching_dedicated, matching_services, dedicated_notes, ) if matching_catch_all: return status( "catch_all", "incomplete", matching_catch_all, matching_services, "Handled by MuseApiContractSupport catch-all, which is not P1R completion.", ) if matching_generic: return status( "generic_persistence", "incomplete", matching_generic, matching_services, "Handled by MuseContractPersistenceService generic persistence, which is not P1R completion.", ) return status("missing", "incomplete", [], matching_services, "No backend route was identified by the conservative scanner.") def status(implementation: str, completion: str, controllers: list[str], services: list[str], notes: str) -> dict[str, Any]: return { "implementationStatus": implementation, "completionStatus": completion, "controllerFiles": controllers, "serviceFiles": services, "notes": notes, } def evidence_test_files_for(domain: str, operation_id: str, completion_status: str) -> list[str]: if completion_status != "completed": return [] if domain == "ai": return list(AI_DOMAIN_TEST_FILES) if domain == "knowledge": return list(KNOWLEDGE_DOMAIN_TEST_FILES) operation_test_files = TEST_FILES_BY_OPERATION.get(operation_key(domain, operation_id)) if operation_test_files is not None: return list(operation_test_files) if domain == "account": return account_test_files_for(operation_id) if domain == "content": return content_test_files_for(operation_id) if domain == "market": return market_test_files_for(operation_id) return [] def account_test_files_for(operation_id: str) -> list[str]: p1r_root = "muse-cloud/muse-server/src/test/java/cn/iocoder/muse/server/framework/api" member_root = "muse-cloud/muse-module-member/muse-module-member-server/src/test/java/cn/iocoder/muse/module/member" mapping = { "adminCreateCallAttributionJob": ( f"{p1r_root}/P1rAccountAdminWritesCompletedApprovalIT.java", f"{member_root}/application/account/AccountAttributionServiceTest.java", ), "adminCreateNewApiBinding": ( f"{p1r_root}/P1rAccountAdminWritesCompletedApprovalIT.java", f"{member_root}/controller/admin/account/AdminAccountNewApiControllerTest.java", ), "adminCreateQuotaAdjustment": ( f"{member_root}/controller/admin/account/AdminAccountUserControllerTest.java", f"{member_root}/application/account/AccountQuotaServiceTest.java", ), "adminCreateQuotaRequest": ( f"{p1r_root}/P1rAccountQuotaRequestCompletedApprovalIT.java", f"{member_root}/application/account/AccountQuotaServiceTest.java", ), "adminGetBalanceSnapshots": ( f"{member_root}/controller/admin/account/AdminAccountUserControllerTest.java", f"{member_root}/application/account/AccountEntitlementServiceTest.java", ), "adminGetCallAttributionJob": ( f"{p1r_root}/P1rAccountAdminReadsCompletedApprovalIT.java", f"{member_root}/application/account/AccountAttributionServiceTest.java", ), "adminGetIntegrationCallByCorrelation": ( f"{p1r_root}/P1rAccountUsageObservabilityCompletedApprovalIT.java", f"{member_root}/application/account/AccountUsageServiceTest.java", ), "adminGetUserEntitlements": ( f"{member_root}/controller/admin/account/AdminAccountUserControllerTest.java", f"{member_root}/application/account/AccountEntitlementServiceTest.java", ), "adminListAccountUsers": ( f"{member_root}/controller/admin/account/AdminAccountUserControllerTest.java", ), "adminListNewApiBindings": ( f"{p1r_root}/P1rAccountNewApiBindingClusterCompletedApprovalIT.java", f"{member_root}/controller/admin/account/AdminAccountNewApiControllerTest.java", ), "adminListPurchaseRecords": ( f"{p1r_root}/P1rAccountAdminPurchaseRecordsCompletedApprovalIT.java", f"{member_root}/controller/admin/account/AdminAccountPurchaseControllerTest.java", ), "adminListQuotaAdjustments": ( f"{member_root}/controller/admin/account/AdminAccountUserControllerTest.java", f"{member_root}/application/account/AccountQuotaServiceTest.java", ), "adminListUsageRecords": ( f"{p1r_root}/P1rAccountAdminReadsCompletedApprovalIT.java", f"{member_root}/controller/admin/account/AdminAccountUsageControllerTest.java", ), "appAcknowledgeSecurityEvent": ( f"{p1r_root}/P1rAccountSecurityEventAckCompletedApprovalIT.java", f"{member_root}/controller/app/account/AppAccountSecurityControllerTest.java", ), "appCreateExportTask": ( f"{p1r_root}/P1rAccountExportDownloadClusterCompletedApprovalIT.java", f"{member_root}/controller/app/account/AppAccountExportControllerTest.java", ), "appCreateQuotaRequest": ( f"{p1r_root}/P1rAccountQuotaRequestCompletedApprovalIT.java", f"{member_root}/application/account/AccountQuotaServiceTest.java", ), "appDownloadExport": ( f"{p1r_root}/P1rAccountExportDownloadClusterCompletedApprovalIT.java", f"{member_root}/controller/app/account/AppAccountExportControllerTest.java", ), "appGetExportTask": ( f"{p1r_root}/P1rAccountExportDownloadClusterCompletedApprovalIT.java", f"{member_root}/controller/app/account/AppAccountExportControllerTest.java", ), "appGetIntegrationCallByCorrelation": ( f"{p1r_root}/P1rAccountUsageObservabilityCompletedApprovalIT.java", f"{member_root}/application/account/AccountUsageServiceTest.java", ), "appGetQuotaRequest": ( f"{p1r_root}/P1rAccountQuotaRequestCompletedApprovalIT.java", f"{member_root}/application/account/AccountQuotaServiceTest.java", ), "appGetSecurityEvent": ( f"{p1r_root}/P1rAccountSecurityEventsCompletedApprovalIT.java", f"{member_root}/controller/app/account/AppAccountSecurityControllerTest.java", ), "appListLicenses": ( f"{p1r_root}/P1rAccountMarketRecordsCompletedApprovalIT.java", f"{member_root}/application/account/AccountMarketRecordServiceTest.java", ), "appListPublishRecords": ( f"{p1r_root}/P1rAccountMarketRecordsCompletedApprovalIT.java", f"{member_root}/application/account/AccountMarketRecordServiceTest.java", ), "appListPurchases": ( f"{p1r_root}/P1rAccountMarketRecordsCompletedApprovalIT.java", f"{member_root}/application/account/AccountMarketRecordServiceTest.java", ), "appListSecurityEvents": ( f"{p1r_root}/P1rAccountSecurityEventsCompletedApprovalIT.java", f"{member_root}/controller/app/account/AppAccountSecurityControllerTest.java", ), "appRecheckNewApiBinding": ( f"{p1r_root}/P1rAccountNewApiBindingClusterCompletedApprovalIT.java", f"{member_root}/controller/app/account/AppAccountNewApiControllerTest.java", ), "getAppBalanceSnapshots": ( f"{member_root}/controller/app/account/AppAccountEntitlementControllerTest.java", f"{member_root}/application/account/AccountEntitlementServiceTest.java", ), "getAppEntitlements": ( f"{member_root}/controller/app/account/AppAccountEntitlementControllerTest.java", f"{member_root}/application/account/AccountEntitlementServiceTest.java", ), "getAppNewApiBinding": ( f"{p1r_root}/P1rAccountNewApiBindingClusterCompletedApprovalIT.java", f"{member_root}/controller/app/account/AppAccountNewApiControllerTest.java", ), "getAppUsage": ( f"{p1r_root}/P1rAccountUsageObservabilityCompletedApprovalIT.java", f"{member_root}/application/account/AccountUsageServiceTest.java", ), "getCurrentUser": ( f"{member_root}/controller/app/account/AppAccountProfileControllerTest.java", f"{member_root}/application/account/AccountProfileServiceTest.java", ), "getProfile": ( f"{member_root}/controller/app/account/AppAccountProfileControllerTest.java", f"{member_root}/application/account/AccountProfileServiceTest.java", ), "updateProfile": ( f"{member_root}/controller/app/account/AppAccountProfileControllerTest.java", f"{member_root}/application/account/AccountProfileServiceTest.java", ), } return list(mapping.get(operation_id, ())) def content_test_files_for(operation_id: str) -> list[str]: p1r_root = "muse-cloud/muse-server/src/test/java/cn/iocoder/muse/server/framework/api" content_root = "muse-cloud/muse-module-content/muse-module-content-server/src/test/java/cn/iocoder/muse/module/content" mapping = { "adminGetWork": (f"{p1r_root}/P1rContentAdminReadCompletedApprovalIT.java",), "adminListChapters": (f"{p1r_root}/P1rContentAdminReadCompletedApprovalIT.java",), "adminListExportTasks": (f"{p1r_root}/P1rContentAdminReadCompletedApprovalIT.java",), "adminListImportTasks": (f"{p1r_root}/P1rContentAdminReadCompletedApprovalIT.java",), "adminListWorks": (f"{p1r_root}/P1rContentAdminReadCompletedApprovalIT.java",), "adminRiskAction": (f"{p1r_root}/P1rContentAdminRiskActionCompletedApprovalIT.java",), "batchConfirmChapters": (f"{p1r_root}/P1rContentParseJobLifecycleCompletedApprovalIT.java",), "confirmChapterParseResult": (f"{p1r_root}/P1rContentChapterParseResultDecisionCompletedApprovalIT.java",), "confirmPlanningCandidate": (f"{p1r_root}/P1rContentPlanningCandidateClusterCompletedApprovalIT.java",), "createBlock": (f"{p1r_root}/P1rContentWorkLifecycleCompletedApprovalIT.java",), "createChapter": (f"{p1r_root}/P1rContentWorkLifecycleCompletedApprovalIT.java",), "createExportTask": (f"{p1r_root}/P1rContentExportTaskCompletedApprovalIT.java",), "createImportTask": (f"{p1r_root}/P1rContentImportTaskCompletedApprovalIT.java",), "createParseJob": (f"{p1r_root}/P1rContentParseJobLifecycleCompletedApprovalIT.java",), "createPlanningCandidate": (f"{p1r_root}/P1rContentPlanningCandidateClusterCompletedApprovalIT.java",), "createStyleCheck": (f"{p1r_root}/P1rContentStyleCheckClusterCompletedApprovalIT.java",), "createWork": (f"{p1r_root}/P1rContentWorkLifecycleCompletedApprovalIT.java",), "deleteBlock": (f"{p1r_root}/P1rContentWorkLifecycleCompletedApprovalIT.java",), "deleteChapter": (f"{p1r_root}/P1rContentWorkLifecycleCompletedApprovalIT.java",), "deleteWork": (f"{p1r_root}/P1rContentWorkLifecycleCompletedApprovalIT.java",), "discardPlanningCandidate": (f"{p1r_root}/P1rContentPlanningCandidateClusterCompletedApprovalIT.java",), "downloadExportPackage": (f"{p1r_root}/P1rContentExportTaskCompletedApprovalIT.java",), "exportWork": (f"{p1r_root}/P1rContentExportTaskCompletedApprovalIT.java",), "getBlock": (f"{p1r_root}/P1rContentCoreCompletedApprovalIT.java",), "getBlockSourceAttribution": (f"{p1r_root}/P1rContentCoreCompletedApprovalIT.java",), "listBlockRevisions": (f"{p1r_root}/P1rContentCoreCompletedApprovalIT.java",), "getChapter": (f"{p1r_root}/P1rContentCoreCompletedApprovalIT.java",), "getExportTask": (f"{p1r_root}/P1rContentExportTaskCompletedApprovalIT.java",), "getImportTask": (f"{p1r_root}/P1rContentImportTaskCompletedApprovalIT.java",), "getMetaProjection": (f"{p1r_root}/P1rContentMetaProjectionClusterCompletedApprovalIT.java",), "getParseJob": (f"{p1r_root}/P1rContentParseJobLifecycleCompletedApprovalIT.java",), "getPlanning": ( f"{p1r_root}/P1rContentPlanningCompletedApprovalIT.java", f"{content_root}/application/ContentPlanningServiceTest.java", ), "getPlanningCandidate": (f"{p1r_root}/P1rContentPlanningCandidateClusterCompletedApprovalIT.java",), "getStyleCheckResult": (f"{p1r_root}/P1rContentStyleCheckClusterCompletedApprovalIT.java",), "getWork": (f"{p1r_root}/P1rContentCoreCompletedApprovalIT.java",), "listBlocks": (f"{p1r_root}/P1rContentCoreCompletedApprovalIT.java",), "listChapters": (f"{p1r_root}/P1rContentCoreCompletedApprovalIT.java",), "listMetaProjections": (f"{p1r_root}/P1rContentMetaProjectionClusterCompletedApprovalIT.java",), "listParseJobChapters": (f"{p1r_root}/P1rContentParseJobLifecycleCompletedApprovalIT.java",), "listPlanningCandidates": (f"{p1r_root}/P1rContentPlanningCandidateClusterCompletedApprovalIT.java",), "listWorks": (f"{p1r_root}/P1rContentCoreCompletedApprovalIT.java",), "mergeBlockSuggestion": (f"{p1r_root}/P1rContentMergeSuggestionIT.java",), "mergeBlocks": (f"{p1r_root}/P1rContentWorkLifecycleCompletedApprovalIT.java",), "rejectChapterParseResult": (f"{p1r_root}/P1rContentChapterParseResultDecisionCompletedApprovalIT.java",), "reorderChapters": (f"{p1r_root}/P1rContentWorkLifecycleCompletedApprovalIT.java",), "retryParseJob": (f"{p1r_root}/P1rContentParseJobLifecycleCompletedApprovalIT.java",), "saveBlock": (f"{p1r_root}/P1rContentCoreCompletedApprovalIT.java",), "savePlanningItem": ( f"{p1r_root}/P1rContentPlanningCompletedApprovalIT.java", f"{content_root}/application/ContentPlanningServiceTest.java", ), "splitBlock": (f"{p1r_root}/P1rContentWorkLifecycleCompletedApprovalIT.java",), "updateChapter": (f"{p1r_root}/P1rContentWorkLifecycleCompletedApprovalIT.java",), "updateWork": (f"{p1r_root}/P1rContentWorkLifecycleCompletedApprovalIT.java",), "validateDynamicFields": (f"{p1r_root}/P1rContentMetaProjectionClusterCompletedApprovalIT.java",), } return list(mapping.get(operation_id, ())) def market_test_files_for(operation_id: str) -> list[str]: p1r_root = "muse-cloud/muse-server/src/test/java/cn/iocoder/muse/server/framework/api" mapping = { "adminApprovePublishRequest": (f"{p1r_root}/P1rMarketAdminPublishReviewCompletedApprovalIT.java",), "adminDelistAsset": (f"{p1r_root}/P1rMarketGovernanceWriteCompletedApprovalIT.java",), "adminGetAppeal": (f"{p1r_root}/P1rMarketAdminAppealCompletedApprovalIT.java",), "adminGetMarketAsset": (f"{p1r_root}/P1rMarketAdminAssetReadsCompletedApprovalIT.java",), "adminListAppeals": (f"{p1r_root}/P1rMarketAdminAppealCompletedApprovalIT.java",), "adminListMarketAssets": (f"{p1r_root}/P1rMarketAdminAssetReadsCompletedApprovalIT.java",), "adminListPublishRequests": (f"{p1r_root}/P1rMarketAdminPublishReviewCompletedApprovalIT.java",), "adminPreviewGovernanceImpact": (f"{p1r_root}/P1rMarketGovernanceWriteCompletedApprovalIT.java",), "adminRecallAsset": (f"{p1r_root}/P1rMarketGovernanceWriteCompletedApprovalIT.java",), "adminRejectPublishRequest": (f"{p1r_root}/P1rMarketAdminPublishReviewCompletedApprovalIT.java",), "adminResolveAppeal": (f"{p1r_root}/P1rMarketAdminAppealCompletedApprovalIT.java",), "cancelHandoff": (f"{p1r_root}/P1rMarketHandoffClusterCompletedApprovalIT.java",), "createBindPrecheck": (f"{p1r_root}/P1rMarketHandoffClusterCompletedApprovalIT.java",), "createMarketplaceHandoff": (f"{p1r_root}/P1rMarketHandoffClusterCompletedApprovalIT.java",), "favoriteAsset": (f"{p1r_root}/P1rMarketDiscoveryFavoriteCompletedApprovalIT.java",), "getGovernanceImpact": (f"{p1r_root}/P1rMarketplaceDiscoveryReadsCompletedApprovalIT.java",), "getHandoffStatus": (f"{p1r_root}/P1rMarketHandoffClusterCompletedApprovalIT.java",), "getMarketplaceAsset": (f"{p1r_root}/P1rMarketDiscoveryFavoriteCompletedApprovalIT.java",), "installMarketplaceAsset": (f"{p1r_root}/P1rMarketLicenseInstallCompletedApprovalIT.java",), "listMarketplaceAssets": (f"{p1r_root}/P1rMarketplaceDiscoveryReadsCompletedApprovalIT.java",), "listMarketplaceCategories": (f"{p1r_root}/P1rMarketDiscoveryFavoriteCompletedApprovalIT.java",), "listMarketplaceRecommendations": (f"{p1r_root}/P1rMarketplaceDiscoveryReadsCompletedApprovalIT.java",), "listMyPublishRecords": (f"{p1r_root}/P1rMarketPublishProducerCompletedApprovalIT.java",), "purchaseAsset": (f"{p1r_root}/P1rMarketLicenseInstallCompletedApprovalIT.java",), "runPublishCheck": (f"{p1r_root}/P1rMarketPublishProducerCompletedApprovalIT.java",), "savePublishDraft": (f"{p1r_root}/P1rMarketPublishProducerCompletedApprovalIT.java",), "submitAppeal": (f"{p1r_root}/P1rMarketProducerAppealCompletedApprovalIT.java",), "submitPublishRequest": (f"{p1r_root}/P1rMarketPublishProducerCompletedApprovalIT.java",), "supplementAppeal": (f"{p1r_root}/P1rMarketProducerAppealCompletedApprovalIT.java",), "unfavoriteAsset": (f"{p1r_root}/P1rMarketDiscoveryFavoriteCompletedApprovalIT.java",), "withdrawAppeal": (f"{p1r_root}/P1rMarketProducerAppealCompletedApprovalIT.java",), "withdrawPublishRequest": (f"{p1r_root}/P1rMarketPublishProducerCompletedApprovalIT.java",), } return list(mapping.get(operation_id, ())) def operation_key(domain: str, operation_id: str) -> str: return f"{domain}:{operation_id}" def dedicated_status_for(domain: str, operation_id: str) -> tuple[str, str]: # AI/Knowledge 是历史 domain-level 用户批准；Events/Meta 使用 operation-level approval， # 避免未来新增 operation 自动继承 completed。 if domain in APPROVED_COMPLETED_DOMAINS or operation_key(domain, operation_id) in APPROVED_COMPLETED_OPERATIONS: return ( "completed", "Dedicated entry exists; user-approved P1R acceptance evidence allows this operation to advance to completed coverage.", ) return ( "needs_verification", "Dedicated entry exists; later P1R stage must verify DTO, state machine, persistence, and real external closure.", ) def files_containing_route_or_operation( files: list[str], text_by_file: dict[str, str], path: str, side: str, method: str, operation_id: str, allow_operation_id_fallback: bool, ) -> list[str]: route_variants = route_variants_for(path) matches: list[str] = [] for file_path in files: if not file_matches_side(file_path, side): continue text = text_by_file[file_path] route_mappings = extract_route_mappings(text) route_matched = any(route_mapping_matches(route_mapping, method, route_variants) for route_mapping in route_mappings) operation_id_matched = allow_operation_id_fallback and operation_id_occurs(text, operation_id) if route_matched or operation_id_matched: matches.append(file_path) return matches def files_containing_sse_placeholder_route(route_by_file: dict[str, list[RouteMapping]], path: str, side: str, method: str) -> list[str]: route_variants = route_variants_for(path) matches: list[str] = [] for file_path, route_mappings in route_by_file.items(): if not file_matches_side(file_path, side): continue if any(route_mapping_matches(route_mapping, method, route_variants) for route_mapping in route_mappings): matches.append(file_path) return matches def file_matches_side(file_path: str, side: str) -> bool: if "/controller/admin/" in file_path: return side == "admin" if "/controller/app/" in file_path: return side == "app" return True def route_variants_for(path: str) -> list[str]: variants = { path, path.removeprefix("/app-api"), path.removeprefix("/admin-api"), } for variant in list(variants): if variant.startswith("/muse/"): variants.add(variant.removeprefix("/muse")) normalized_variants = {variant if variant.startswith("/") else "/" + variant for variant in variants} return sorted(normalized_variants, key=len, reverse=True) def extract_route_literals(text: str) -> list[str]: return re.findall(r'"(/[^"]*)"', text) def extract_route_mappings(text: str) -> list[RouteMapping]: class_routes = extract_class_request_mapping_routes(text) mappings: list[RouteMapping] = [] for match in re.finditer(r"@(Get|Post|Put|Patch|Delete|Request)Mapping(?:\s*\((.*?)\))?", text, flags=re.DOTALL): if is_before_primary_type_declaration(text, match.start()): continue method_mappings = route_mappings_from_annotation(match.group(1), match.group(2) or "", allow_empty_route=True) mappings.extend(combine_class_and_method_routes(class_routes, method_mappings)) return mappings def extract_class_request_mapping_routes(text: str) -> tuple[str, ...]: primary_type_index = primary_type_declaration_index(text) if primary_type_index == -1: return ("",) routes: set[str] = set() for match in re.finditer(r"@RequestMapping(?:\s*\((.*?)\))?", text[:primary_type_index], flags=re.DOTALL): routes.update(extract_route_literals(match.group(1) or "")) return tuple(sorted(routes)) if routes else ("",) def is_before_primary_type_declaration(text: str, index: int) -> bool: primary_type_index = primary_type_declaration_index(text) return primary_type_index != -1 and index < primary_type_index def primary_type_declaration_index(text: str) -> int: match = re.search(r"\b(?:class|interface|record)\s+[A-Za-z_][A-Za-z0-9_]*", text) return match.start() if match else -1 def route_mappings_from_annotation(annotation_kind: str, annotation_body: str, allow_empty_route: bool = False) -> list[RouteMapping]: routes = tuple(sorted(set(extract_route_literals(annotation_body)))) if not routes: if not allow_empty_route: return [] routes = ("",) if annotation_kind in MAPPING_METHODS: return [RouteMapping(methods=(MAPPING_METHODS[annotation_kind],), routes=routes)] request_methods = tuple(sorted(set(re.findall(r"RequestMethod\.([A-Z]+)", annotation_body)) & JAVA_HTTP_METHODS)) if not request_methods: return [] return [RouteMapping(methods=request_methods, routes=routes)] def combine_class_and_method_routes(class_routes: tuple[str, ...], method_mappings: list[RouteMapping]) -> list[RouteMapping]: combined_mappings: list[RouteMapping] = [] for method_mapping in method_mappings: routes = tuple( sorted( { join_route_paths(class_route, method_route) for class_route in class_routes for method_route in method_mapping.routes } ) ) combined_mappings.append(RouteMapping(methods=method_mapping.methods, routes=routes)) return combined_mappings def join_route_paths(class_route: str, method_route: str) -> str: # Spring MVC 会把类级 base path 与方法级相对 path 合并；空方法路径表示使用类级路径本身。 parts = [part.strip("/") for part in (class_route, method_route) if part and part.strip("/")] return "/" + "/".join(parts) if parts else "/" def route_mapping_matches(route_mapping: RouteMapping, method: str, route_variants: list[str]) -> bool: if method not in route_mapping.methods: return False return any( route_literal_matches(route_literal, route_variant) for route_literal in route_mapping.routes for route_variant in route_variants ) def route_literal_matches(route_literal: str, route_variant: str) -> bool: literal = normalize_route_template(route_literal) variant = normalize_route_template(route_variant) if literal == variant: return True if literal.endswith("/**"): base = literal.removesuffix("/**") return variant == base or variant.startswith(base + "/") return False def normalize_route_template(route: str) -> str: # Java Controller 与 OpenAPI 都用 {name} 表示路径变量；变量名不同也应视为同一路由形状。 return re.sub(r"\{[^}/]+\}", "{}", route) def operation_id_occurs(text: str, operation_id: str) -> bool: # 只允许标识符边界匹配，避免 createWork 命中 createWorkflow 这类前缀误判。 return re.search(rf"(? list[str]: domain_path = "/module/" + ("member" if domain == "account" else domain) + "/" return [file_path for file_path in files if domain_path in file_path] def normalize_path(path: str) -> str: if path.startswith("/app-api/") or path.startswith("/admin-api/"): return path return path def detect_side(path: str) -> str: if path.startswith("/admin-api/"): return "admin" if path.startswith("/app-api/"): return "app" return "unknown" def operation_requires_command_id(operation_doc: dict[str, Any], document: dict[str, Any]) -> bool: if operation_has_required_command_header(operation_doc, document): return True # 只把 schema.required 中的 commandId 视为必填；properties 或 description 里的 commandId 不代表请求必须携带。 for content_doc in (operation_doc.get("requestBody") or {}).get("content", {}).values(): if not isinstance(content_doc, dict): continue schema_doc = content_doc.get("schema") if isinstance(schema_doc, dict) and local_schema_requires_command_id(schema_doc, document, set()): return True return False def operation_has_required_command_header(operation_doc: dict[str, Any], document: dict[str, Any]) -> bool: for parameter_doc in operation_doc.get("parameters") or []: if not isinstance(parameter_doc, dict): continue resolved = resolve_parameter(parameter_doc, document) if not isinstance(resolved, dict): continue if resolved.get("in") == "header" and resolved.get("name") == "X-Command-Id" and resolved.get("required") is True: return True return False def resolve_parameter(parameter_doc: dict[str, Any], document: dict[str, Any]) -> Any: ref = parameter_doc.get("$ref") if isinstance(ref, str) and ref.startswith("#/"): return resolve_local_ref(document, ref) return parameter_doc def local_schema_requires_command_id(schema_doc: dict[str, Any], document: dict[str, Any], seen_refs: set[str]) -> bool: required_fields = schema_doc.get("required") or [] if "commandId" in required_fields: return True ref = schema_doc.get("$ref") if isinstance(ref, str) and ref.startswith("#/") and ref not in seen_refs: seen_refs.add(ref) resolved = resolve_local_ref(document, ref) if isinstance(resolved, dict) and local_schema_requires_command_id(resolved, document, seen_refs): return True # 只递归 OpenAPI 组合 schema；不递归 properties，避免把可选 commandId 误判成必填。 for key in ("allOf", "anyOf", "oneOf"): for nested_schema in schema_doc.get(key) or []: if isinstance(nested_schema, dict) and local_schema_requires_command_id(nested_schema, document, seen_refs): return True return False def resolve_local_ref(document: dict[str, Any], ref: str) -> Any: current: Any = document for part in ref.removeprefix("#/").split("/"): if not isinstance(current, dict): return None current = current.get(part) return current def is_sse_placeholder_controller(text: str) -> bool: if "SseEmitter" not in text: return False if any(marker in text for marker in SSE_PLACEHOLDER_READY_MARKERS): return True if ".name(\"done\")" in text: return True return "stream ready" in text and "emitter.send(SseEmitter.event()" in text and "emitter.complete();" in text def sse_placeholder_route_mappings(text: str) -> list[RouteMapping]: if not is_sse_placeholder_controller(text): return [] mappings: set[RouteMapping] = set() markers = list(SSE_PLACEHOLDER_READY_MARKERS) + [".name(\"done\")", "stream ready"] for marker in markers: search_from = 0 while True: marker_index = text.find(marker, search_from) if marker_index == -1: break annotation_index = nearest_mapping_annotation_before(text, marker_index) if annotation_index != -1: for match in re.finditer( r"@(Get|Post|Put|Patch|Delete|Request)Mapping(?:\s*\((.*?)\))?", text[annotation_index:marker_index], flags=re.DOTALL, ): mappings.update(route_mappings_from_annotation(match.group(1), match.group(2) or "")) search_from = marker_index + len(marker) return sorted(mappings, key=lambda mapping: (mapping.methods, mapping.routes)) def nearest_mapping_annotation_before(text: str, marker_index: int) -> int: mapping_annotations = ["@GetMapping", "@PostMapping", "@PutMapping", "@PatchMapping", "@DeleteMapping", "@RequestMapping"] return max(text.rfind(annotation, 0, marker_index) for annotation in mapping_annotations) def detect_external_dependencies(domain: str, path: str, operation_id: str) -> list[str]: haystack = f"{domain} {path} {operation_id}".lower() dependencies: set[str] = set() if any(word in haystack for word in ["ai", "task", "suggestion", "agent", "job", "source-event"]): dependencies.update(["New-API", "SSE"]) if any(word in haystack for word in ["knowledge", "kb", "document", "entity", "relation", "graph", "reindex"]): dependencies.add("RAGFlow") if any(word in haystack for word in ["import", "export", "download", "file", "document"]): dependencies.add("FileService") if any(word in haystack for word in ["marketplace", "purchase", "install", "handoff", "license"]): dependencies.update(["Account", "MarketAuthorization"]) if any(word in haystack for word in ["account", "usage", "quota", "balance", "correlation", "new-api-binding"]): dependencies.update(["New-API", "AccountLedger"]) return sorted(dependencies) def target_stage(domain: str) -> str: return { "content": "P1R-1 Content Real API", "meta": "P1R-2 Meta Real API", "account": "P1R-3 Account Real API", "ai": "P1R-4 AI Real API", "knowledge": "P1R-5 Knowledge Real API", "market": "P1R-6 Market Real API", "events": "P1R-7 End-to-End Acceptance", }.get(domain, "P1R-0 Baseline Gate") def build_report(operations: list[Operation]) -> dict[str, Any]: summary = { "totalOperations": len(operations), "completedOperations": sum(1 for operation in operations if operation.completionStatus == "completed"), "incompleteOperations": sum(1 for operation in operations if operation.completionStatus == "incomplete"), "needsVerificationOperations": sum(1 for operation in operations if operation.completionStatus == "needs_verification"), "catchAllOperations": sum(1 for operation in operations if operation.implementationStatus == "catch_all"), "genericPersistenceOperations": sum(1 for operation in operations if operation.implementationStatus == "generic_persistence"), "ssePlaceholderOperations": sum(1 for operation in operations if operation.implementationStatus == "sse_placeholder"), "missingOperations": sum(1 for operation in operations if operation.implementationStatus == "missing"), "blockedOperations": sum(1 for operation in operations if operation.completionStatus == "blocked"), } return { "generatedAt": stable_generated_at(operations), "sourceContracts": [str(path.relative_to(REPO_ROOT)) for path in sorted(CONTRACT_ROOT.glob("*/openapi.yaml"))], "summary": summary, "operations": [asdict(operation) for operation in operations], } def stable_generated_at(operations: list[Operation]) -> str: # 让提交的基线报告可重复生成，避免每次 --check 只因为当前时间产生无意义 diff。 payload = json.dumps([asdict(operation) for operation in operations], ensure_ascii=False, sort_keys=True) digest_seconds = int(hashlib.sha256(payload.encode("utf-8")).hexdigest()[:8], 16) % 86400 hours, remainder = divmod(digest_seconds, 3600) minutes, seconds = divmod(remainder, 60) return f"2026-05-25T{hours:02d}:{minutes:02d}:{seconds:02d}+00:00" def render_markdown(report: dict[str, Any]) -> str: lines = [ "# P1R API Coverage Report", "", f"- Generated at: `{report['generatedAt']}`", f"- Total operations: `{report['summary']['totalOperations']}`", f"- Completed: `{report['summary']['completedOperations']}`", f"- Needs verification: `{report['summary']['needsVerificationOperations']}`", f"- Incomplete: `{report['summary']['incompleteOperations']}`", f"- Catch-all: `{report['summary']['catchAllOperations']}`", f"- Generic persistence: `{report['summary']['genericPersistenceOperations']}`", f"- SSE placeholder: `{report['summary']['ssePlaceholderOperations']}`", f"- Missing: `{report['summary']['missingOperations']}`", "", "## Operations", "", "| Domain | Side | Method | Path | Operation | Implementation | Completion | Target Stage |", "|--------|------|--------|------|-----------|----------------|------------|--------------|", ] for operation in report["operations"]: lines.append( "| {domain} | {side} | {method} | `{path}` | `{operationId}` | {implementationStatus} | {completionStatus} | {targetStage} |".format( **operation ) ) lines.extend(["", "## Incomplete Or Blocked", ""]) for operation in report["operations"]: if operation["completionStatus"] in {"incomplete", "blocked"}: lines.append( f"- `{operation['operationId']}` `{operation['method']} {operation['path']}`: " f"{operation['implementationStatus']} - {operation['notes']}" ) lines.append("") return "\n".join(lines) def validate_report(report: dict[str, Any]) -> None: required_fields = { "domain", "side", "method", "path", "operationId", "isWrite", "requiresCommandId", "externalDependencies", "targetStage", "implementationStatus", "completionStatus", "controllerFiles", "serviceFiles", "testFiles", "notes", } if report["summary"]["totalOperations"] <= 0: raise SystemExit("No operations found") for operation in report["operations"]: missing = required_fields - operation.keys() if missing: raise SystemExit(f"Operation {operation.get('operationId')} missing fields: {sorted(missing)}") if ( operation["implementationStatus"] in NON_REAL_IMPLEMENTATION_STATUSES and operation["completionStatus"] not in NON_REAL_COMPLETION_STATUSES ): raise SystemExit( f"{operation['operationId']} non-real implementation must be incomplete or blocked" ) if operation["completionStatus"] == "completed" and not approved_completed_operation(operation): raise SystemExit(f"{operation['operationId']} is incorrectly marked completed") if operation["completionStatus"] == "completed" and not operation["testFiles"]: raise SystemExit(f"{operation['operationId']} completed operation must reference testFiles evidence") if operation["completionStatus"] != "completed" and operation["testFiles"]: raise SystemExit(f"{operation['operationId']} unfinished operation must not reference testFiles") for test_file in operation["testFiles"]: validate_test_file(operation, test_file) if "message" in json.dumps(operation, ensure_ascii=False): raise SystemExit(f"{operation['operationId']} contains forbidden response evidence field: message") def validate_test_file(operation: dict[str, Any], test_file: str) -> None: if "/src/test/java/" not in test_file or not test_file.endswith(".java"): raise SystemExit(f"{operation['operationId']} testFiles must point to Java test source: {test_file}") if "P1rApiCoverageReportTest.java" in test_file or "RealApiGateTest.java" in test_file: raise SystemExit(f"{operation['operationId']} testFiles must not self-reference coverage gates: {test_file}") if not (REPO_ROOT / test_file).exists(): raise SystemExit(f"{operation['operationId']} testFiles path does not exist: {test_file}") def approved_completed_operation(operation: dict[str, Any]) -> bool: # completed 只能来自已批准推进的 dedicated 域或单项 operation，避免非目标 domain 被误计完成。 if operation["implementationStatus"] != "dedicated": return False if operation["domain"] in APPROVED_COMPLETED_DOMAINS: return True return operation_key(operation["domain"], operation["operationId"]) in APPROVED_COMPLETED_OPERATIONS if __name__ == "__main__": raise SystemExit(main())