from __future__ import annotations from copy import deepcopy import json from typing import Any from uuid import uuid4 from mcp.server.fastmcp import FastMCP from pydantic import BaseModel, ValidationError from ..config import DEFAULT_PROFILE from ..errors import QingflowApiError, backend_code_int, is_auth_like_error, raise_tool_error from ..list_type_labels import get_record_list_type_label from ..solution.build_assembly_store import BuildAssemblyStore, default_manifest from ..solution.compiler import CompiledSolution, ExecutionPlan, ExecutionStep, build_execution_plan, compile_solution from ..solution.design_session import DesignStage, evaluate_design_session, finalize_design_session, merge_design_payload from ..solution.design_store import DesignSessionStore from ..solution.executor import SolutionExecutor from ..solution.normalizer import normalize_solution_spec from ..solution.requirements_builder import GeneratedAppBuild, RequirementsBuildError, build_app_spec_from_requirements from ..solution.run_store import RunArtifactStore, fingerprint_payload from ..solution.spec_models import ( AnalyticsPortalBuildSpec, AppBuildSpec, AppFlowBuildSpec, FlowBuildSpec, BuildManifest, NavigationBuildSpec, SolutionSpec, ViewsBuildSpec, ) from .app_tools import AppTools from .base import ToolBase from .navigation_tools import NavigationTools from .package_tools import PackageTools from .portal_tools import PortalTools from .qingbi_report_tools import QingbiReportTools from .record_tools import RecordTools from .role_tools import RoleTools from .view_tools import ViewTools from .workspace_tools import WorkspaceTools STAGED_BUILD_MODES = {"preflight", "plan", "apply", "repair"} STAGED_BUILD_MODE_ALIASES = { "create": "plan", "new": "plan", "start": "plan", "draft": "plan", "update": "plan", "modify": "plan", "preview": "preflight", "prepare": "preflight", "run": "apply", "execute": "apply", "fix": "repair", "resume": "repair", "retry": "repair", } STAGE_ORDER = ["app_flow", "views", "analytics_portal", "navigation"] STAGE_TOOL_NAMES = { "app_flow": "solution_build_app_flow", "views": "solution_build_views", "analytics_portal": "solution_build_analytics_portal", "navigation": "solution_build_navigation", } SOLUTION_SCHEMA_STAGES = {"app", "app_update", "flow", "views", "analytics_portal", "navigation", "app_flow", "all"} SOLUTION_SCHEMA_INTENTS = {"minimal", "full", "demo"} SOLUTION_SCHEMA_STAGE_ALIASES = { "all": "all", "solution": "all", "full_solution": "all", "app": "app", "application": "app", "form": "app", "schema": "app", "app_spec": "app", "app_update": "app_update", "update_app": "app_update", "update": "app_update", "更新应用": "app_update", "更新表单": "app_update", "更新schema": "app_update", "应用": "app", "表单": "app", "flow": "flow", "workflow": "flow", "流程": "flow", "flow_spec": "flow", "views": "views", "view": "views", "视图": "views", "analytics_portal": "analytics_portal", "analytics": "analytics_portal", "portal": "analytics_portal", "report": "analytics_portal", "dashboard": "analytics_portal", "报表": "analytics_portal", "门户": "analytics_portal", "navigation": "navigation", "nav": "navigation", "menu": "navigation", "导航": "navigation", "app_flow": "app_flow", "应用流程": "app_flow", } SOLUTION_SCHEMA_INTENT_ALIASES = { "minimal": "minimal", "min": "minimal", "最小": "minimal", "最简": "minimal", "骨架": "minimal", "基础": "minimal", "full": "full", "complete": "full", "完整": "full", "全字段": "full", "完整布局": "full", "all_fields": "full", "demo": "demo", "example": "demo", "sample": "demo", "示例": "demo", "演示": "demo", } class SolutionTools(ToolBase): """解决方案工具（中文名：方案会话与安装执行）。 类型：解决方案编排工具。 主要职责： 1. 管理设计会话与运行会话； 2. 安装与应用 SolutionSpec； 3. 输出安装结果、阶段状态与故障信息。 """ def register(self, mcp: FastMCP) -> None: """注册当前工具到 MCP 服务。""" @mcp.tool() def solution_design_session( action: str = "get", session_id: str = "", stage: str | None = None, design_patch: dict[str, Any] | None = None, metadata: dict[str, Any] | None = None, ) -> dict[str, Any]: return self.solution_design_session( action=action, session_id=session_id, stage=stage, design_patch=design_patch or {}, metadata=metadata or {}, ) @mcp.tool() def solution_build_all( profile: str = DEFAULT_PROFILE, mode: str = "plan", build_id: str = "", solution_spec: dict[str, Any] | None = None, publish: bool = True, run_label: str | None = None, target: dict[str, Any] | None = None, repair_patch: dict[str, Any] | None = None, verify: bool = False, ) -> dict[str, Any]: return self.solution_build_all( profile=profile, mode=mode, build_id=build_id, solution_spec=solution_spec or {}, publish=publish, run_label=run_label, target=target or {}, repair_patch=repair_patch or {}, verify=verify, ) @mcp.tool() def solution_build_app( profile: str = DEFAULT_PROFILE, mode: str = "plan", build_id: str = "", app_spec: dict[str, Any] | None = None, app_key: str = "", package_tag_id: int = 0, update_only: bool = False, publish: bool = True, run_label: str | None = None, target: dict[str, Any] | None = None, repair_patch: dict[str, Any] | None = None, ) -> dict[str, Any]: return self.solution_build_app( profile=profile, mode=mode, build_id=build_id, app_spec=app_spec or {}, app_key=app_key, package_tag_id=package_tag_id, update_only=update_only, publish=publish, run_label=run_label, target=target or {}, repair_patch=repair_patch or {}, ) @mcp.tool() def solution_build_app_from_requirements( profile: str = DEFAULT_PROFILE, mode: str = "plan", build_id: str = "", title: str = "", requirement_text: str = "", package_tag_id: int = 0, package_name: str = "", app_key: str = "", update_only: bool = False, layout_style: str = "auto", publish: bool = True, run_label: str | None = None, include_generated_spec: bool = False, ) -> dict[str, Any]: return self.solution_build_app_from_requirements( profile=profile, mode=mode, build_id=build_id, title=title, requirement_text=requirement_text, package_tag_id=package_tag_id, package_name=package_name, app_key=app_key, update_only=update_only, layout_style=layout_style, publish=publish, run_label=run_label, include_generated_spec=include_generated_spec, ) @mcp.tool() def solution_build_flow( profile: str = DEFAULT_PROFILE, mode: str = "plan", build_id: str = "", flow_spec: dict[str, Any] | None = None, publish: bool = True, run_label: str | None = None, repair_patch: dict[str, Any] | None = None, ) -> dict[str, Any]: return self.solution_build_flow( profile=profile, mode=mode, build_id=build_id, flow_spec=flow_spec or {}, publish=publish, run_label=run_label, repair_patch=repair_patch or {}, ) @mcp.tool() def solution_build_app_flow( profile: str = DEFAULT_PROFILE, mode: str = "plan", build_id: str = "", app_flow_spec: dict[str, Any] | None = None, publish: bool = True, run_label: str | None = None, target: dict[str, Any] | None = None, repair_patch: dict[str, Any] | None = None, ) -> dict[str, Any]: return self.solution_build_app_flow( profile=profile, mode=mode, build_id=build_id, app_flow_spec=app_flow_spec or {}, publish=publish, run_label=run_label, target=target or {}, repair_patch=repair_patch or {}, ) @mcp.tool() def solution_schema_example( stage: str = "app", intent: str = "minimal", include_examples: bool = False, include_selected_example: bool = False, ) -> dict[str, Any]: return self.solution_schema_example( stage=stage, intent=intent, include_examples=include_examples, include_selected_example=include_selected_example, ) @mcp.tool() def solution_build_views( profile: str = DEFAULT_PROFILE, mode: str = "plan", build_id: str = "", views_spec: dict[str, Any] | None = None, publish: bool = True, run_label: str | None = None, repair_patch: dict[str, Any] | None = None, ) -> dict[str, Any]: return self.solution_build_views( profile=profile, mode=mode, build_id=build_id, views_spec=views_spec or {}, publish=publish, run_label=run_label, repair_patch=repair_patch or {}, ) @mcp.tool() def solution_build_analytics_portal( profile: str = DEFAULT_PROFILE, mode: str = "plan", build_id: str = "", analytics_portal_spec: dict[str, Any] | None = None, publish: bool = True, run_label: str | None = None, repair_patch: dict[str, Any] | None = None, ) -> dict[str, Any]: return self.solution_build_analytics_portal( profile=profile, mode=mode, build_id=build_id, analytics_portal_spec=analytics_portal_spec or {}, publish=publish, run_label=run_label, repair_patch=repair_patch or {}, ) @mcp.tool() def solution_build_navigation( profile: str = DEFAULT_PROFILE, mode: str = "plan", build_id: str = "", navigation_spec: dict[str, Any] | None = None, publish: bool = True, run_label: str | None = None, repair_patch: dict[str, Any] | None = None, ) -> dict[str, Any]: return self.solution_build_navigation( profile=profile, mode=mode, build_id=build_id, navigation_spec=navigation_spec or {}, publish=publish, run_label=run_label, repair_patch=repair_patch or {}, ) @mcp.tool() def solution_build_status(build_id: str = "") -> dict[str, Any]: return self.solution_build_status(build_id=build_id) def solution_design_session( self, *, action: str, session_id: str, stage: str | None, design_patch: dict[str, Any], metadata: dict[str, Any], ) -> dict[str, Any]: """执行方案相关逻辑。""" try: if action == "start" and not session_id: session_id = _generate_session_id(metadata=metadata, stage=stage) if action not in {"start", "get", "submit_stage", "finalize"}: return self._design_failure_response(session_id, action, "config", "action must be one of: start, get, submit_stage, finalize") if not session_id: return self._design_failure_response(session_id, action, "config", "session_id is required") store = DesignSessionStore.open(session_id=session_id, metadata=metadata) if metadata: store.set_metadata(metadata) if action == "start": evaluation = evaluate_design_session(store.data.get("stage_payloads", {})) store.update_progress( status=evaluation["status"], current_stage=evaluation["current_stage"], stage_results=evaluation["stage_results"], merged_design_spec=evaluation["merged_design_spec"], ) return self._design_summary(store) if action == "get": return self._design_summary(store) if action == "submit_stage": if stage not in {item.value for item in DesignStage if item != DesignStage.finalize}: return self._design_failure_response(session_id, action, "config", "stage must be one of: discover, design, experience") if not isinstance(design_patch, dict) or not design_patch: return self._design_failure_response(session_id, action, "config", "design_patch must be a non-empty object") existing_payload = store.get_stage_payload(stage) merged_payload = merge_design_payload(existing_payload, design_patch) store.set_stage_payload(stage, merged_payload) evaluation = evaluate_design_session(store.data.get("stage_payloads", {})) store.update_progress( status=evaluation["status"], current_stage=evaluation["current_stage"], stage_results=evaluation["stage_results"], merged_design_spec=evaluation["merged_design_spec"], ) return self._design_summary(store) finalized = finalize_design_session(store.data.get("stage_payloads", {})) store.update_progress( status=finalized["status"], current_stage=finalized["current_stage"], stage_results=finalized["stage_results"], merged_design_spec=finalized["merged_design_spec"], ) store.mark_finalized( normalized_solution_spec=finalized["normalized_solution_spec"], execution_plan=finalized["execution_plan"], ) return self._design_summary(store) except ValidationError as exc: return self._design_failure_response(session_id, action, "validation", exc.errors()) except ValueError as exc: return self._design_failure_response(session_id, action, "config", str(exc)) except Exception as exc: # noqa: BLE001 return self._design_failure_response(session_id, action, "runtime", str(exc)) def solution_bootstrap( self, *, profile: str, mode: str, solution_spec: dict[str, Any], idempotency_key: str, publish: bool, run_label: str | None, target: dict[str, Any], repair_patch: dict[str, Any], ) -> dict[str, Any]: """执行方案相关逻辑。""" mode = _normalize_staged_build_mode(mode) if mode not in {"preflight", "plan", "apply", "repair"}: return self._failure_response(mode, idempotency_key, "config", "mode must be one of: preflight, plan, apply, repair") if not idempotency_key: return self._failure_response(mode, idempotency_key, "config", "idempotency_key is required") if not isinstance(solution_spec, dict) or not solution_spec: return self._failure_response(mode, idempotency_key, "config", "solution_spec must be a non-empty object") if mode != "repair" and repair_patch: return self._failure_response(mode, idempotency_key, "config", "repair_patch is only supported in repair mode") def runner() -> dict[str, Any]: try: base_spec_payload = deepcopy(solution_spec) if mode == "repair" and repair_patch: base_spec_payload = deep_merge(base_spec_payload, repair_patch) parsed = SolutionSpec.model_validate(base_spec_payload) normalized = normalize_solution_spec(parsed) compiled = compile_solution(normalized) package_tag_id = _resolve_target_package_tag_id(target) if package_tag_id is not None: compiled = _bind_existing_package_target(compiled, package_tag_id) request_fingerprint = fingerprint_payload( { "mode": "apply" if mode == "repair" else mode, "publish": publish, "solution_spec": normalized.model_dump(mode="json"), "target": target, } ) if mode in {"preflight", "plan"}: store = BuildAssemblyStore.open(build_id=idempotency_key) store.set_manifest(normalized.model_dump(mode="json")) return { "build_id": idempotency_key, "mode": mode, "idempotency_key": idempotency_key, "normalized_solution_spec": normalized.model_dump(mode="json"), "execution_plan": compiled.execution_plan.as_dict(), "artifacts": {}, "step_results": {}, "publish_results": {}, "status": "preflighted" if mode == "preflight" else "planned", "errors": [], "run_path": None, "build_path": str(store.path), "suggested_next_call": _solution_build_next_call( tool_name="solution_build_all", mode=mode, build_id=idempotency_key, ), } store = RunArtifactStore.open( idempotency_key=idempotency_key, normalized_solution_spec=normalized.model_dump(mode="json"), request_fingerprint=request_fingerprint, run_label=run_label, ) if mode == "apply": store.ensure_apply_fingerprint(request_fingerprint) if package_tag_id is not None: store.set_artifact("package", "tag_id", package_tag_id) store.set_artifact("package", "reused", True) result = self._build_executor().execute(profile=profile, compiled=compiled, store=store, publish=publish, mode=mode) return { "mode": mode, "idempotency_key": idempotency_key, "normalized_solution_spec": normalized.model_dump(mode="json"), "execution_plan": compiled.execution_plan.as_dict(), "artifacts": result["artifacts"], "step_results": result["step_results"], "publish_results": _extract_publish_results(result["step_results"]), "status": result["status"], "errors": result["errors"], "run_path": result["run_path"], } except ValidationError as exc: return self._failure_response(mode, idempotency_key, "validation", exc.errors()) except ValueError as exc: return self._failure_response(mode, idempotency_key, "config", str(exc)) except Exception as exc: # noqa: BLE001 return self._failure_response(mode, idempotency_key, "runtime", str(exc)) if mode in {"preflight", "plan"}: return runner() return self._run(profile, lambda _session_profile, _context: runner(), tool_name='方案启动搭建') def solution_build_app_flow( self, *, profile: str, mode: str, build_id: str, app_flow_spec: dict[str, Any], publish: bool, run_label: str | None, target: dict[str, Any], repair_patch: dict[str, Any], ) -> dict[str, Any]: """执行方案相关逻辑。""" mode = _normalize_staged_build_mode(mode) return self._stage_build( profile=profile, mode=mode, build_id=build_id, stage_name="app_flow", stage_payload=app_flow_spec, stage_model=AppFlowBuildSpec, publish=publish, run_label=run_label, target=target, repair_patch=repair_patch, ) def solution_build_app( self, *, profile: str, mode: str, build_id: str, app_spec: dict[str, Any], app_key: str = "", package_tag_id: int = 0, update_only: bool = False, publish: bool, run_label: str | None, target: dict[str, Any], repair_patch: dict[str, Any], ) -> dict[str, Any]: """执行方案相关逻辑。""" mode = _normalize_staged_build_mode(mode) resolved_app_spec = deepcopy(app_spec) if isinstance(app_spec, dict) else {} if not resolved_app_spec and build_id: resolved_app_spec = self._load_generated_app_spec(build_id) or {} resolved_target = _merge_app_target( target, app_key=app_key, package_tag_id=package_tag_id, update_only=update_only, ) return self._stage_build( profile=profile, mode=mode, build_id=build_id, stage_name="app_flow", stage_payload=resolved_app_spec, stage_model=AppBuildSpec, publish=publish, run_label=run_label, target=resolved_target, repair_patch=repair_patch, public_stage_name="app", tool_name="solution_build_app", projection_name="app", ) def solution_build_app_from_requirements( self, *, profile: str, mode: str, build_id: str, title: str, requirement_text: str, package_tag_id: int, package_name: str, app_key: str = "", update_only: bool = False, layout_style: str, publish: bool, run_label: str | None, include_generated_spec: bool, ) -> dict[str, Any]: """执行方案相关逻辑。""" mode = _normalize_staged_build_mode(mode) if mode in {"preflight", "plan"} and not build_id: build_id = _generate_build_id(run_label=run_label, stage_name="app") resolved_target = _merge_app_target( {}, app_key=app_key, package_tag_id=package_tag_id, update_only=update_only, ) resolved_package = self._resolve_builder_package_reference( profile=profile, package_tag_id=package_tag_id, package_name=package_name, ) if resolved_package.get("status") == "failed": return resolved_package["response"] existing_app = self._resolve_builder_existing_app_reference( profile=profile, app_key=app_key, package_resolution=resolved_package, update_only=_target_is_update_only(resolved_target), ) if existing_app.get("status") == "failed": return existing_app["response"] effective_package_resolution = existing_app.get("package_resolution") if existing_app.get("package_resolution") else resolved_package try: generated = build_app_spec_from_requirements( title=title, requirement_text=requirement_text, package_name=effective_package_resolution.get("tag_name"), layout_style=layout_style, ) except RequirementsBuildError as exc: return self._requirements_failure_response( mode=mode, build_id=build_id, title=title, requirement_text=requirement_text, package_resolution=effective_package_resolution, app_key=app_key, update_only=_target_is_update_only(resolved_target), layout_style=layout_style, publish=publish, run_label=run_label, detail=str(exc), error_code=exc.error_code, recoverable=exc.recoverable, missing_required_fields=exc.missing_required_fields, invalid_field_types=exc.invalid_field_types, error_details=exc.details, ) except ValueError as exc: return self._requirements_failure_response( mode=mode, build_id=build_id, title=title, requirement_text=requirement_text, package_resolution=effective_package_resolution, app_key=app_key, update_only=_target_is_update_only(resolved_target), layout_style=layout_style, publish=publish, run_label=run_label, detail=str(exc), ) if build_id: self._persist_generated_app_spec( build_id=build_id, generated=generated, package_resolution=effective_package_resolution, title=title, requirement_text=requirement_text, layout_style=layout_style, ) stage_result = self.solution_build_app( profile=profile, mode=mode, build_id=build_id, app_spec=generated.app_spec, app_key=app_key, package_tag_id=int(effective_package_resolution.get("tag_id") or 0), update_only=_target_is_update_only(resolved_target), publish=publish, run_label=run_label, target=_requirements_target( effective_package_resolution, app_key=app_key if isinstance(app_key, str) else "", update_only=_target_is_update_only(resolved_target), ), repair_patch={}, ) return self._requirements_success_response( mode=mode, title=title, requirement_text=requirement_text, package_resolution=effective_package_resolution, layout_style=layout_style, publish=publish, run_label=run_label, generated=generated, stage_result=stage_result, include_generated_spec=include_generated_spec, ) def solution_build_flow( self, *, profile: str, mode: str, build_id: str, flow_spec: dict[str, Any], publish: bool, run_label: str | None, repair_patch: dict[str, Any], ) -> dict[str, Any]: """执行方案相关逻辑。""" mode = _normalize_staged_build_mode(mode) app_key = str(flow_spec.get("app_key") or "").strip() spec_payload = flow_spec.get("spec") if app_key and isinstance(spec_payload, dict) and spec_payload: from .ai_builder_tools import AiBuilderTools builder = AiBuilderTools(self.sessions, self.backend) if mode in {"preflight", "plan"}: return { "build_id": build_id or _generate_build_id(run_label=run_label, stage_name="flow"), "mode": mode, "stage": "flow", "status": "planned" if mode == "plan" else "preflighted", "normalized_args": {"app_key": app_key, "spec": spec_payload}, "tool_name": "solution_build_flow", } result = builder.app_flow_apply( profile=profile, app_key=app_key, spec=spec_payload, publish=publish, schema_version=flow_spec.get("schema_version") or flow_spec.get("schemaVersion"), ) return { "build_id": build_id, "mode": mode, "stage": "flow", "status": result.get("status"), "result": result, "tool_name": "solution_build_flow", } return self._stage_build( profile=profile, mode=mode, build_id=build_id, stage_name="app_flow", stage_payload=flow_spec, stage_model=FlowBuildSpec, publish=publish, run_label=run_label, target={}, repair_patch=repair_patch, public_stage_name="flow", tool_name="solution_build_flow", projection_name="flow", ) def solution_build_all( self, *, profile: str, mode: str, build_id: str, solution_spec: dict[str, Any], publish: bool, run_label: str | None, target: dict[str, Any], repair_patch: dict[str, Any], verify: bool, ) -> dict[str, Any]: """执行方案相关逻辑。""" mode = _normalize_staged_build_mode(mode) if mode in {"preflight", "plan"} and not build_id: build_id = _generate_build_id(run_label=run_label, stage_name="all") if mode not in STAGED_BUILD_MODES: return self._all_failure_response(build_id, mode, "config", "mode must be one of: preflight, plan, apply, repair") if not build_id: return self._all_failure_response(build_id, mode, "config", "build_id is required") if not isinstance(solution_spec, dict): return self._all_failure_response(build_id, mode, "config", "solution_spec must be an object") if mode != "repair" and not solution_spec: return self._all_failure_response(build_id, mode, "config", "solution_spec must be a non-empty object") if mode != "repair" and repair_patch: return self._all_failure_response(build_id, mode, "config", "repair_patch is only supported in repair mode") def runner() -> dict[str, Any]: try: base_solution_spec = deepcopy(solution_spec) if mode == "repair" and not base_solution_spec: assembly = BuildAssemblyStore.open(build_id=build_id, create=False) base_solution_spec = assembly.get_manifest() if not base_solution_spec.get("entities"): return self._all_failure_response( build_id, mode, "config", "repair requires a stored build manifest or a non-empty solution_spec", ) resolved_solution_spec = self._resolve_full_solution_spec( solution_spec=base_solution_spec, mode=mode, repair_patch=repair_patch, ) stage_payloads = _split_solution_spec_into_stage_payloads(resolved_solution_spec) stage_results: dict[str, Any] = {} overall_errors: list[dict[str, Any]] = [] if mode == "preflight": overall_status = "preflighted" elif mode == "plan": overall_status = "planned" else: overall_status = "success" for stage_name in STAGE_ORDER: stage_payload = stage_payloads[stage_name] if _stage_payload_is_empty(stage_name, stage_payload): assembly = BuildAssemblyStore.open(build_id=build_id) assembly.add_stage_history( { "stage": stage_name, "mode": mode, "status": "skipped", "reason": _stage_skip_reason(stage_name), "execution_step_count": 0, } ) stage_results[stage_name] = self._stage_skip_response( build_id=build_id, stage_name=stage_name, mode=mode, assembly=assembly, tool_name=STAGE_TOOL_NAMES.get(stage_name, "solution_build_app_flow"), ) continue if stage_name == "app_flow": result = self._stage_build( profile=profile, mode=mode, build_id=build_id, stage_name=stage_name, stage_payload=stage_payload, stage_model=AppFlowBuildSpec, publish=publish, run_label=run_label, target=target, repair_patch={}, ) elif stage_name == "views": result = self._stage_build( profile=profile, mode=mode, build_id=build_id, stage_name=stage_name, stage_payload=stage_payload, stage_model=ViewsBuildSpec, publish=publish, run_label=run_label, target={}, repair_patch={}, ) elif stage_name == "analytics_portal": result = self._stage_build( profile=profile, mode=mode, build_id=build_id, stage_name=stage_name, stage_payload=stage_payload, stage_model=AnalyticsPortalBuildSpec, publish=publish, run_label=run_label, target={}, repair_patch={}, ) else: result = self._stage_build( profile=profile, mode=mode, build_id=build_id, stage_name=stage_name, stage_payload=stage_payload, stage_model=NavigationBuildSpec, publish=publish, run_label=run_label, target={}, repair_patch={}, ) stage_results[stage_name] = result if result.get("status") == "failed": overall_status = "failed" overall_errors = _sanitize_errors(result.get("errors", [])) break if mode in {"apply", "repair"} and result.get("status") != "success": overall_status = str(result.get("status")) elif mode == "plan" and result.get("status") != "planned": overall_status = str(result.get("status")) elif mode == "preflight" and result.get("status") != "preflighted": overall_status = str(result.get("status")) assembly = BuildAssemblyStore.open(build_id=build_id) if overall_status != "failed" and mode in {"apply", "repair"}: assembly.mark_status("success") build_summary = _build_summary(assembly) artifacts = assembly.get_artifacts() verification = None if verify and overall_status == "success": verification = self._verify_solution_build( profile=profile, build_id=build_id, artifacts=artifacts, solution_spec=resolved_solution_spec, ) return { "build_id": build_id, "mode": mode, "normalized_solution_spec": resolved_solution_spec, "stage_results": stage_results, "stage_statuses": {stage: str(result.get("status")) for stage, result in stage_results.items()}, "artifacts": artifacts, "status": overall_status, "errors": overall_errors, "verification": verification, "build_summary": build_summary, "build_path": build_summary.get("build_path"), "suggested_next_call": _solution_build_next_call(tool_name="solution_build_all", mode=mode, build_id=build_id), } except ValidationError as exc: return self._all_failure_response(build_id, mode, "validation", exc.errors()) except FileNotFoundError: return self._all_failure_response( build_id, mode, "config", "repair requires an existing build_id with a stored build manifest", ) except ValueError as exc: return self._all_failure_response(build_id, mode, "config", str(exc)) except Exception as exc: # noqa: BLE001 return self._all_failure_response(build_id, mode, "runtime", str(exc)) if mode in {"preflight", "plan"}: return runner() return self._run(profile, lambda _session_profile, _context: runner(), tool_name='一键搭建全部') def solution_build_views( self, *, profile: str, mode: str, build_id: str, views_spec: dict[str, Any], publish: bool, run_label: str | None, repair_patch: dict[str, Any], ) -> dict[str, Any]: """执行方案相关逻辑。""" mode = _normalize_staged_build_mode(mode) return self._stage_build( profile=profile, mode=mode, build_id=build_id, stage_name="views", stage_payload=views_spec, stage_model=ViewsBuildSpec, publish=publish, run_label=run_label, target={}, repair_patch=repair_patch, ) def solution_schema_example( self, *, stage: str, intent: str, include_examples: bool = False, include_selected_example: bool = False, ) -> dict[str, Any]: """执行方案相关逻辑。""" resolved_stage = _normalize_solution_schema_stage(stage) resolved_intent = _normalize_solution_schema_intent(intent) if resolved_stage is None: return { "stage": stage, "intent": intent, "status": "failed", "errors": [{"category": "config", "detail": f"stage must be one of: {', '.join(sorted(SOLUTION_SCHEMA_STAGES))}"}], "available_stages": sorted(SOLUTION_SCHEMA_STAGES), **_solution_error_fields( category="config", detail=f"stage must be one of: {', '.join(sorted(SOLUTION_SCHEMA_STAGES))}", suggested_next_call={"tool_name": "solution_schema_example", "arguments": {"stage": "app", "intent": "minimal"}}, ), } if resolved_intent is None: return { "stage": resolved_stage, "intent": intent, "status": "failed", "errors": [{"category": "config", "detail": f"intent must be one of: {', '.join(sorted(SOLUTION_SCHEMA_INTENTS))}"}], "available_intents": sorted(SOLUTION_SCHEMA_INTENTS), **_solution_error_fields( category="config", detail=f"intent must be one of: {', '.join(sorted(SOLUTION_SCHEMA_INTENTS))}", suggested_next_call={"tool_name": "solution_schema_example", "arguments": {"stage": resolved_stage, "intent": "minimal"}}, ), } response = _solution_schema_example_payload( stage=resolved_stage, intent=resolved_intent, include_examples=include_examples, include_selected_example=include_selected_example, ) response["requested_stage"] = stage response["requested_intent"] = intent response["resolved_stage"] = resolved_stage response["resolved_intent"] = resolved_intent return response def solution_build_analytics_portal( self, *, profile: str, mode: str, build_id: str, analytics_portal_spec: dict[str, Any], publish: bool, run_label: str | None, repair_patch: dict[str, Any], ) -> dict[str, Any]: """执行方案相关逻辑。""" mode = _normalize_staged_build_mode(mode) return self._stage_build( profile=profile, mode=mode, build_id=build_id, stage_name="analytics_portal", stage_payload=analytics_portal_spec, stage_model=AnalyticsPortalBuildSpec, publish=publish, run_label=run_label, target={}, repair_patch=repair_patch, ) def solution_build_navigation( self, *, profile: str, mode: str, build_id: str, navigation_spec: dict[str, Any], publish: bool, run_label: str | None, repair_patch: dict[str, Any], ) -> dict[str, Any]: """执行方案相关逻辑。""" mode = _normalize_staged_build_mode(mode) return self._stage_build( profile=profile, mode=mode, build_id=build_id, stage_name="navigation", stage_payload=navigation_spec, stage_model=NavigationBuildSpec, publish=publish, run_label=run_label, target={}, repair_patch=repair_patch, ) def solution_build_status(self, *, build_id: str) -> dict[str, Any]: """执行方案相关逻辑。""" if not build_id: error_fields = _solution_error_fields( category="config", detail="build_id is required", suggested_next_call=None, ) return { "build_id": build_id, "status": "failed", "errors": [{"category": "config", "detail": "build_id is required"}], "build_path": None, **error_fields, } assembly = BuildAssemblyStore.open(build_id=build_id) summary = _build_summary(assembly) summary["errors"] = [] return summary def _stage_build( self, *, profile: str, mode: str, build_id: str, stage_name: str, stage_payload: dict[str, Any], stage_model: type[BaseModel], publish: bool, run_label: str | None, target: dict[str, Any], repair_patch: dict[str, Any], public_stage_name: str | None = None, tool_name: str | None = None, projection_name: str | None = None, ) -> dict[str, Any]: """执行内部辅助逻辑。""" mode = _normalize_staged_build_mode(mode) public_stage_name = public_stage_name or stage_name tool_name = tool_name or STAGE_TOOL_NAMES.get(stage_name, "solution_build_app_flow") projection_name = projection_name or public_stage_name if mode in {"preflight", "plan"} and not build_id: build_id = _generate_build_id(run_label=run_label, stage_name=stage_name) assembly = BuildAssemblyStore.open(build_id=build_id) if build_id else None if mode not in STAGED_BUILD_MODES: return self._stage_failure_response(build_id, public_stage_name, mode, "config", "mode must be one of: preflight, plan, apply, repair", assembly, tool_name=tool_name) if not build_id: return self._stage_failure_response(build_id, public_stage_name, mode, "config", "build_id is required", assembly, tool_name=tool_name) if mode != "repair" and repair_patch: return self._stage_failure_response(build_id, public_stage_name, mode, "config", "repair_patch is only supported in repair mode", assembly, tool_name=tool_name) if stage_name != "app_flow" and target: return self._stage_failure_response(build_id, public_stage_name, mode, "config", "target is only supported for the app_flow stage", assembly, tool_name=tool_name) def runner() -> dict[str, Any]: return self._execute_stage_build( profile=profile, build_id=build_id, stage_name=stage_name, public_stage_name=public_stage_name, mode=mode, assembly=assembly, stage_payload=stage_payload, stage_model=stage_model, publish=publish, run_label=run_label, target=target, repair_patch=repair_patch, tool_name=tool_name, projection_name=projection_name, ) if mode in {"preflight", "plan"}: return runner() return self._run(profile, lambda _session_profile, _context: runner(), tool_name=tool_name) def _execute_stage_build( self, *, profile: str, build_id: str, stage_name: str, public_stage_name: str, mode: str, assembly: BuildAssemblyStore, stage_payload: dict[str, Any], stage_model: type[BaseModel], publish: bool, run_label: str | None, target: dict[str, Any], repair_patch: dict[str, Any], tool_name: str, projection_name: str, ) -> dict[str, Any]: """执行内部辅助逻辑。""" try: resolved_stage_payload = self._resolve_stage_payload( assembly=assembly, stage_name=stage_name, mode=mode, incoming_stage_payload=stage_payload, repair_patch=repair_patch, stage_model=stage_model, projection_name=projection_name, ) missing_prerequisites = _missing_stage_prerequisites( stage_name=public_stage_name, assembly=assembly, stage_payload=resolved_stage_payload, ) if missing_prerequisites: raise ValueError("; ".join(missing_prerequisites)) merged_manifest = _merge_stage_manifest( stage_name=stage_name, manifest=assembly.get_manifest(), stage_payload=resolved_stage_payload, ) parsed_manifest = BuildManifest.model_validate(merged_manifest) normalized_manifest = normalize_solution_spec(parsed_manifest) compiled = compile_solution(normalized_manifest) existing_app_targets = _resolve_target_app_keys( target=target, entity_ids=[entity.entity_id for entity in compiled.entities], ) update_only = _target_is_update_only(target) or bool(existing_app_targets) if update_only and not existing_app_targets: raise ValueError("update target requires app_key or target.app_keys") package_tag_id = _resolve_target_package_tag_id(target) if package_tag_id is not None: compiled = _bind_existing_package_target(compiled, package_tag_id) elif update_only: compiled = _bind_existing_app_target(compiled) filtered_compiled = _filter_compiled_solution( compiled, stage_name=stage_name, stage_payload=resolved_stage_payload, stage_variant=public_stage_name, ) assembly.set_stage_spec(stage_name, resolved_stage_payload) assembly.set_manifest(normalized_manifest.model_dump(mode="json")) if existing_app_targets: _seed_existing_app_artifacts(assembly=assembly, existing_app_targets=existing_app_targets) if mode in {"preflight", "plan"}: status = "preflighted" if mode == "preflight" else "planned" assembly.add_stage_history( { "stage": stage_name, "mode": mode, "status": status, "execution_step_count": len(filtered_compiled.execution_plan.steps), } ) assembly.mark_status("draft") return self._stage_success_response( build_id=build_id, stage_name=public_stage_name, mode=mode, normalized_manifest=normalized_manifest, compiled=filtered_compiled, artifacts=assembly.get_artifacts(), step_results={}, errors=[], run_path=None, build_path=str(assembly.path), status=status, build_summary=_build_summary(assembly), tool_name=tool_name, ) request_fingerprint = fingerprint_payload( { "stage": stage_name, "mode": "apply" if mode == "repair" else mode, "publish": publish, "build_manifest": normalized_manifest.model_dump(mode="json"), "stage_payload": resolved_stage_payload, "target": target, } ) store = RunArtifactStore.open( idempotency_key=_stage_run_id(build_id, stage_name), normalized_solution_spec=normalized_manifest.model_dump(mode="json"), request_fingerprint=request_fingerprint, run_label=run_label, initial_artifacts=assembly.get_artifacts(), ) if mode == "apply": store.ensure_apply_fingerprint(request_fingerprint) if package_tag_id is not None: store.set_artifact("package", "tag_id", package_tag_id) store.set_artifact("package", "reused", True) result = self._build_executor().execute( profile=profile, compiled=filtered_compiled, store=store, publish=publish, mode=mode, ) assembly.set_artifacts(result["artifacts"]) stage_verification = None effective_status = result["status"] if result["status"] == "success": stage_verification = self._verify_stage_write_result( profile=profile, build_id=build_id, public_stage_name=public_stage_name, artifacts=result["artifacts"], solution_spec=normalized_manifest.model_dump(mode="json"), target=target, ) if isinstance(stage_verification, dict) and stage_verification.get("status") == "partial": effective_status = "partial_success" failure_signature = None stage_failure_count = 0 next_strategy = None suggested_next_call_override = None if effective_status == "failed": failure_signature, stage_failure_count = self._record_failure_signature( assembly=assembly, stage_name=public_stage_name, stage_payload=resolved_stage_payload, errors=result.get("errors", []), ) if failure_signature and stage_failure_count >= 2 and public_stage_name == "app": next_strategy = "switch_to_explicit_app_stage" suggested_next_call_override = { "tool_name": "solution_build_app", "arguments": {"mode": "plan", "build_id": build_id}, } assembly.add_stage_history( { "stage": stage_name, "mode": mode, "status": effective_status, "run_path": result["run_path"], "execution_step_count": len(filtered_compiled.execution_plan.steps), } ) assembly.mark_status(_build_status_for_stage(stage_name, effective_status)) response = self._stage_success_response( build_id=build_id, stage_name=public_stage_name, mode=mode, normalized_manifest=normalized_manifest, compiled=filtered_compiled, artifacts=result["artifacts"], step_results=result["step_results"], errors=result["errors"], run_path=result["run_path"], build_path=str(assembly.path), status=effective_status, build_summary=_build_summary(assembly), tool_name=tool_name, verification=stage_verification, ) if effective_status == "failed": response["failure_signature"] = failure_signature response["stage_failure_count"] = stage_failure_count or 1 if next_strategy: response["next_strategy"] = next_strategy if suggested_next_call_override is not None: response["suggested_next_call"] = suggested_next_call_override return response except ValidationError as exc: return self._stage_failure_response(build_id, public_stage_name, mode, "validation", exc.errors(), assembly, tool_name=tool_name) except ValueError as exc: return self._stage_failure_response(build_id, public_stage_name, mode, "config", str(exc), assembly, tool_name=tool_name) except Exception as exc: # noqa: BLE001 return self._stage_failure_response(build_id, public_stage_name, mode, "runtime", str(exc), assembly, tool_name=tool_name) def _resolve_stage_payload( self, *, assembly: BuildAssemblyStore, stage_name: str, mode: str, incoming_stage_payload: dict[str, Any], repair_patch: dict[str, Any], stage_model: type[BaseModel], projection_name: str, ) -> dict[str, Any]: """执行内部辅助逻辑。""" existing_stage_payload = (assembly.data.get("stage_specs", {}) or {}).get(stage_name, {}) or {} base_payload: dict[str, Any] = {} if mode == "repair": base_payload = deepcopy(existing_stage_payload) if incoming_stage_payload: base_payload = deep_merge(base_payload, incoming_stage_payload) if repair_patch: base_payload = deep_merge(base_payload, repair_patch) if not base_payload: raise ValueError(f"{stage_name} stage does not have a stored spec to repair") else: if incoming_stage_payload: base_payload = deepcopy(incoming_stage_payload) elif existing_stage_payload: base_payload = deepcopy(existing_stage_payload) else: raise ValueError(f"{stage_name} stage spec is required") parsed: BaseModel | None = None try: parsed = stage_model.model_validate(base_payload) except ValidationError as original_error: if _looks_like_full_solution_spec(base_payload): parsed = stage_model.model_validate(_project_stage_payload(stage_name=projection_name, solution_spec=base_payload)) else: raise original_error resolved = parsed.model_dump(mode="json", exclude_unset=True) if not resolved: raise ValueError(f"{stage_name} stage spec is required") return resolved def _resolve_full_solution_spec( self, *, solution_spec: dict[str, Any], mode: str, repair_patch: dict[str, Any], ) -> dict[str, Any]: """执行内部辅助逻辑。""" base_payload = deepcopy(solution_spec) if mode == "repair" and repair_patch: base_payload = deep_merge(base_payload, repair_patch) parsed = SolutionSpec.model_validate(base_payload) return parsed.model_dump(mode="json", exclude_unset=True) def _build_executor(self) -> SolutionExecutor: """执行内部辅助逻辑。""" return SolutionExecutor( workspace_tools=WorkspaceTools(self.sessions, self.backend), package_tools=PackageTools(self.sessions, self.backend), role_tools=RoleTools(self.sessions, self.backend), app_tools=AppTools(self.sessions, self.backend), record_tools=RecordTools(self.sessions, self.backend), view_tools=ViewTools(self.sessions, self.backend), chart_tools=QingbiReportTools(self.sessions, self.backend), portal_tools=PortalTools(self.sessions, self.backend), navigation_tools=NavigationTools(self.sessions, self.backend), ) def _verify_solution_build( self, *, profile: str, build_id: str, artifacts: dict[str, Any], solution_spec: dict[str, Any], ) -> dict[str, Any]: """执行内部辅助逻辑。""" verification = { "build_id": build_id, "status": "success", "errors": [], "package": {}, "apps": [], "portal": {}, "navigation": {}, } package_tools = PackageTools(self.sessions, self.backend) app_tools = AppTools(self.sessions, self.backend) portal_tools = PortalTools(self.sessions, self.backend) navigation_tools = NavigationTools(self.sessions, self.backend) record_tools = RecordTools(self.sessions, self.backend) entity_specs = { entity.get("entity_id"): entity for entity in solution_spec.get("entities", []) if isinstance(entity, dict) and isinstance(entity.get("entity_id"), str) } field_maps_artifact = artifacts.get("field_maps", {}) if isinstance(artifacts.get("field_maps"), dict) else {} try: package_artifact = artifacts.get("package", {}) if isinstance(artifacts.get("package"), dict) else {} tag_id = package_artifact.get("tag_id") if isinstance(tag_id, int) and tag_id > 0: package_detail = package_tools.package_get(profile=profile, tag_id=tag_id, include_raw=True).get("result") or {} verification["package"] = { "tag_id": tag_id, "name": package_detail.get("tagName"), "item_count": len(package_detail.get("tagItems") or []), } except Exception as exc: # noqa: BLE001 _append_verification_error(verification, "package", exc) apps_artifact = artifacts.get("apps", {}) if isinstance(artifacts.get("apps"), dict) else {} records_artifact = artifacts.get("records", {}) if isinstance(artifacts.get("records"), dict) else {} for entity_id, app_info in apps_artifact.items(): try: if not isinstance(app_info, dict): continue app_key = app_info.get("app_key") if not isinstance(app_key, str) or not app_key: continue entity_spec = entity_specs.get(entity_id) field_meta = field_maps_artifact.get(entity_id, {}) if isinstance(field_maps_artifact.get(entity_id), dict) else {} base_info = app_tools.app_get_base(profile=profile, app_key=app_key, include_raw=True).get("result") or {} record_page = _readback_record_page( record_tools=record_tools, profile=profile, app_key=app_key, entity_spec=entity_spec, field_meta=field_meta, ) rows = record_page.get("list") or [] returned_rows = _coerce_count(record_page.get("returned_rows")) if returned_rows is None: returned_rows = len(rows) if isinstance(rows, list) else 0 backend_reported_total = _coerce_count(record_page.get("backend_reported_total")) effective_count = _coerce_count(record_page.get("effective_count")) if effective_count is None: effective_count = max(backend_reported_total or 0, returned_rows) seeded_records = records_artifact.get(entity_id, {}) if isinstance(records_artifact.get(entity_id), dict) else {} app_verification = { "entity_id": entity_id, "app_key": app_key, "app_name": base_info.get("formTitle") or app_info.get("display_name") or app_info.get("name"), "app_icon": base_info.get("appIcon"), "returned_rows": returned_rows, "effective_count": effective_count, "list_type": _coerce_count(record_page.get("list_type")), "list_type_label": record_page.get("list_type_label"), "seeded_records": len(seeded_records), "sample_titles": _extract_record_titles(rows, entity_spec=entity_spec, field_meta=field_meta), } if backend_reported_total is not None: app_verification["backend_reported_total"] = backend_reported_total verification["apps"].append(app_verification) except Exception as exc: # noqa: BLE001 _append_verification_error(verification, f"app:{entity_id}", exc) try: portal_artifact = artifacts.get("portal", {}) if isinstance(artifacts.get("portal"), dict) else {} dash_key = portal_artifact.get("dash_key") if isinstance(dash_key, str) and dash_key: published = portal_tools.portal_get(profile=profile, dash_key=dash_key, being_draft=False).get("result") or {} draft = portal_tools.portal_get(profile=profile, dash_key=dash_key, being_draft=True).get("result") or {} verification["portal"] = { "dash_key": dash_key, "name": published.get("dashName") or draft.get("dashName") or published.get("name") or draft.get("name"), "published_component_count": _count_portal_components(published), "draft_component_count": _count_portal_components(draft), } except Exception as exc: # noqa: BLE001 _append_verification_error(verification, "portal", exc) try: navigation_artifact = artifacts.get("navigation", {}) if isinstance(artifacts.get("navigation"), dict) else {} status = navigation_tools.navigation_get_status(profile=profile).get("result") or {} items = navigation_artifact.get("items") or [] verification["navigation"] = { "navigation_id": status.get("navigationId") or status.get("id"), "publish_status": status.get("publishStatus"), "item_count": len(items) if isinstance(items, list) else 0, "items": _extract_navigation_items(items), } except Exception as exc: # noqa: BLE001 _append_verification_error(verification, "navigation", exc) if verification["errors"]: verification["status"] = "partial" return verification def _verify_stage_write_result( self, *, profile: str, build_id: str, public_stage_name: str, artifacts: dict[str, Any], solution_spec: dict[str, Any], target: dict[str, Any], ) -> dict[str, Any] | None: """执行内部辅助逻辑。""" if public_stage_name != "app": return None app_tools = AppTools(self.sessions, self.backend) expected_package_tag_id = _resolve_target_package_tag_id(target) verification: dict[str, Any] = { "stage": public_stage_name, "status": "success", "expected_package_tag_id": expected_package_tag_id, "package_attached": None if expected_package_tag_id is None else True, "apps": [], "views_created": [], "views_strategy": "not_included", "errors": [], } apps_artifact = artifacts.get("apps", {}) if isinstance(artifacts.get("apps"), dict) else {} views_artifact = artifacts.get("views", {}) if isinstance(artifacts.get("views"), dict) else {} entity_specs = { entity.get("entity_id"): entity for entity in solution_spec.get("entities", []) if isinstance(entity, dict) and isinstance(entity.get("entity_id"), str) } for entity_id, app_info in apps_artifact.items(): if not isinstance(app_info, dict): continue app_key = app_info.get("app_key") if not isinstance(app_key, str) or not app_key: continue try: base_info = app_tools.app_get_base(profile=profile, app_key=app_key, include_raw=True).get("result") or {} raw_tag_ids = base_info.get("tagIds") if isinstance(base_info.get("tagIds"), list) else [] tag_ids_after = [tag_id for tag_id in (_coerce_count(value) for value in raw_tag_ids) if tag_id is not None] package_attached = None if expected_package_tag_id is None else expected_package_tag_id in tag_ids_after if package_attached is False: verification["status"] = "partial" verification["package_attached"] = False verification["errors"].append( { "category": "verification", "detail": f"app '{app_key}' is not attached to expected package '{expected_package_tag_id}'", "entity_id": entity_id, "app_key": app_key, "tag_ids_after": tag_ids_after, } ) entity_views = views_artifact.get(entity_id) if isinstance(views_artifact.get(entity_id), dict) else {} created_view_keys = sorted( { str(view_result.get("view_key")) for view_result in entity_views.values() if isinstance(view_result, dict) and isinstance(view_result.get("view_key"), str) and view_result.get("view_key") } ) verification["views_created"].extend(created_view_keys) verification["apps"].append( { "entity_id": entity_id, "display_name": entity_specs.get(entity_id, {}).get("display_name"), "app_key": app_key, "tag_ids_after": tag_ids_after, "package_attached": package_attached, "views_created": created_view_keys, "publish_status": base_info.get("appPublishStatus"), } ) except Exception as exc: # noqa: BLE001 verification["status"] = "partial" error_payload = _verification_error_payload(exc) error_payload.update( { "category": "verification", "entity_id": entity_id, "app_key": app_key, } ) verification["errors"].append(error_payload) if verification["views_created"]: verification["views_strategy"] = "created" else: verification["suggested_next_call"] = { "tool_name": "solution_build_views", "arguments": {"mode": "plan", "build_id": build_id}, } return verification def _failure_response(self, mode: str, idempotency_key: str, category: str, detail: Any) -> dict[str, Any]: """执行内部辅助逻辑。""" error_fields = _solution_error_fields( category=category, detail=detail, suggested_next_call=None, stage="all", ) return { "mode": mode, "idempotency_key": idempotency_key, "normalized_solution_spec": None, "execution_plan": {"steps": []}, "artifacts": {}, "step_results": {}, "publish_results": {}, "status": "failed", "errors": _sanitize_errors([{"category": category, "detail": detail}]), "run_path": None, "verification": None, **error_fields, } def _stage_failure_response( self, build_id: str, stage_name: str, mode: str, category: str, detail: Any, assembly: BuildAssemblyStore | None, *, tool_name: str, ) -> dict[str, Any]: """执行内部辅助逻辑。""" error_fields = _solution_error_fields( category=category, detail=detail, suggested_next_call=_solution_build_next_call( tool_name=tool_name, mode=mode, build_id=build_id or None, ), stage=stage_name, ) if build_id and stage_name == "app": error_fields["suggested_schema_call"] = None return { "build_id": build_id, "stage": stage_name, "mode": mode, "normalized_build_manifest": None, "execution_plan": {"steps": []}, "artifacts": assembly.get_artifacts() if assembly is not None else {}, "step_results": {}, "publish_results": {}, "status": "failed", "errors": _sanitize_errors([{"category": category, "detail": detail}]), "build_path": str(assembly.path) if assembly is not None else None, "run_path": None, "build_summary": _build_summary(assembly) if assembly is not None else None, **error_fields, } def _stage_skip_response( self, *, build_id: str, stage_name: str, mode: str, assembly: BuildAssemblyStore, tool_name: str, ) -> dict[str, Any]: """执行内部辅助逻辑。""" return { "build_id": build_id, "stage": stage_name, "mode": mode, "normalized_build_manifest": assembly.get_manifest(), "execution_plan": {"steps": []}, "artifacts": assembly.get_artifacts(), "step_results": {}, "publish_results": {}, "status": "skipped", "errors": [], "build_path": str(assembly.path), "run_path": None, "build_summary": _build_summary(assembly), "suggested_next_call": _solution_build_next_call( tool_name=tool_name, mode=mode, build_id=build_id, ), } def _stage_success_response( self, *, build_id: str, stage_name: str, mode: str, normalized_manifest: SolutionSpec, compiled: CompiledSolution, artifacts: dict[str, Any], step_results: dict[str, Any], errors: list[dict[str, Any]], run_path: str | None, build_path: str, status: str, build_summary: dict[str, Any], tool_name: str, verification: dict[str, Any] | None = None, ) -> dict[str, Any]: """执行内部辅助逻辑。""" response = { "build_id": build_id, "stage": stage_name, "mode": mode, "normalized_build_manifest": normalized_manifest.model_dump(mode="json"), "execution_plan": compiled.execution_plan.as_dict(), "artifacts": artifacts, "step_results": step_results, "publish_results": _extract_publish_results(step_results), "status": status, "errors": _sanitize_errors(errors), "build_path": build_path, "run_path": run_path, "verification": verification, "build_summary": build_summary, "suggested_next_call": _solution_build_next_call( tool_name=tool_name, mode=mode, build_id=build_id, ), } if isinstance(verification, dict): for key in ("package_attached", "views_created", "views_strategy"): if key in verification: response[key] = verification.get(key) return response def _design_failure_response(self, session_id: str, action: str, category: str, detail: Any) -> dict[str, Any]: """执行内部辅助逻辑。""" error_fields = _solution_error_fields( category=category, detail=detail, suggested_next_call=_design_session_next_call(action=action, session_id=session_id or None), ) return { "session_id": session_id, "status": "failed", "current_stage": None, "next_stage": None, "metadata": {}, "stage_results": {}, "merged_design_spec": {}, "normalized_solution_spec": None, "execution_plan": None, "errors": _sanitize_errors([{"category": category, "detail": detail}]), "session_path": None, **error_fields, } def _design_summary(self, store: DesignSessionStore) -> dict[str, Any]: """执行内部辅助逻辑。""" summary = store.summary() current_stage = summary["current_stage"] summary["next_stage"] = None if summary["status"] == "finalized" else current_stage summary["errors"] = [] summary["suggested_next_call"] = _design_summary_next_call(summary) return summary def _all_failure_response(self, build_id: str, mode: str, category: str, detail: Any) -> dict[str, Any]: """执行内部辅助逻辑。""" error_fields = _solution_error_fields( category=category, detail=detail, suggested_next_call=_solution_build_next_call(tool_name="solution_build_all", mode=mode, build_id=build_id or None), stage="all", ) return { "build_id": build_id, "mode": mode, "normalized_solution_spec": None, "stage_results": {}, "stage_statuses": {}, "artifacts": {}, "status": "failed", "errors": _sanitize_errors([{"category": category, "detail": detail}]), "verification": None, "build_summary": None, "build_path": None, **error_fields, } def _load_generated_app_spec(self, build_id: str) -> dict[str, Any] | None: """执行内部辅助逻辑。""" if not build_id: return None try: assembly = BuildAssemblyStore.open(build_id=build_id, create=False) except FileNotFoundError: return None generated_specs = assembly.get_builder_value("generated_specs", {}) if not isinstance(generated_specs, dict): return None app_spec = generated_specs.get("app") return deepcopy(app_spec) if isinstance(app_spec, dict) else None def _persist_generated_app_spec( self, *, build_id: str, generated: GeneratedAppBuild, package_resolution: dict[str, Any], title: str, requirement_text: str, layout_style: str, ) -> None: """执行内部辅助逻辑。""" assembly = BuildAssemblyStore.open(build_id=build_id) builder_state = assembly.get_builder_state() generated_specs = builder_state.get("generated_specs", {}) if not isinstance(generated_specs, dict): generated_specs = {} generated_specs["app"] = deepcopy(generated.app_spec) generated_specs["app_summary"] = deepcopy(generated.summary) generated_specs["requirements_context"] = { "title": title, "requirement_text": requirement_text, "layout_style": layout_style, "package_resolution": deepcopy(package_resolution), } builder_state["generated_specs"] = generated_specs assembly.set_builder_state(builder_state) def _record_failure_signature( self, *, assembly: BuildAssemblyStore, stage_name: str, stage_payload: dict[str, Any], errors: list[dict[str, Any]], ) -> tuple[str | None, int]: """执行内部辅助逻辑。""" signature = _failure_signature_for_stage(stage_name=stage_name, stage_payload=stage_payload, errors=errors) if signature is None: return None, 0 builder_state = assembly.get_builder_state() signature_counts = builder_state.get("failure_signatures", {}) if not isinstance(signature_counts, dict): signature_counts = {} count = int(signature_counts.get(signature, 0)) + 1 signature_counts[signature] = count builder_state["failure_signatures"] = signature_counts builder_state["last_failure_signature"] = signature builder_state["last_failure_stage"] = stage_name builder_state["last_failure_count"] = count assembly.set_builder_state(builder_state) return signature, count def _resolve_builder_package_reference( self, *, profile: str, package_tag_id: int, package_name: str, ) -> dict[str, Any]: """执行内部辅助逻辑。""" packages = PackageTools(self.sessions, self.backend) normalized_name = package_name.strip() if package_tag_id > 0: try: result = packages.package_get(profile=profile, tag_id=package_tag_id, include_raw=False) except (QingflowApiError, RuntimeError) as exc: error = _coerce_solution_api_error(exc) if is_auth_like_error(error): raise_tool_error(error) if backend_code_int(error) in {40002, 40027}: return { "status": "resolved", "matched_via": "tag_id", "tag_id": package_tag_id, "tag_name": None, "candidates": [], "metadata_unverified": True, "lookup_permission_blocked": { "scope": "package", "target": {"tag_id": package_tag_id}, "transport_error": { "http_status": error.http_status, "backend_code": error.backend_code, "category": error.category, "request_id": error.request_id, }, }, } return _builder_package_resolution_failed( package_name=normalized_name, package_tag_id=package_tag_id, error=error, retried=False, ) summary = result.get("result") if isinstance(result.get("result"), dict) else {} resolution: dict[str, Any] = { "status": "resolved", "matched_via": "tag_id", "tag_id": package_tag_id, "tag_name": summary.get("tagName"), "candidates": [], } warnings = result.get("warnings") if isinstance(warnings, list): for warning in warnings: if not isinstance(warning, dict): continue if warning.get("code") not in {"PACKAGE_BASE_INFO_UNAVAILABLE", "PACKAGE_DETAIL_READ_DEGRADED"}: continue resolution["metadata_unverified"] = True resolution["lookup_permission_blocked"] = { "scope": "package", "target": {"tag_id": package_tag_id}, "transport_error": { "http_status": warning.get("http_status"), "backend_code": warning.get("backend_code"), "category": "backend", "request_id": warning.get("request_id"), }, } break return resolution if not normalized_name: return { "status": "new_package", "matched_via": "none", "tag_id": None, "tag_name": None, "candidates": [], "source_shape": None, "retried": False, } try: result = packages.package_list(profile=profile, trial_status="all", include_raw=False) except (QingflowApiError, RuntimeError) as exc: return _builder_package_resolution_failed( package_name=normalized_name, package_tag_id=package_tag_id, error=_coerce_solution_api_error(exc), retried=False, ) items = result.get("items") if isinstance(result.get("items"), list) else [] source_shape = result.get("source_shape") retried = False if not items: try: retry_result = packages.package_list(profile=profile, trial_status="all", include_raw=False) except (QingflowApiError, RuntimeError) as exc: return _builder_package_resolution_failed( package_name=normalized_name, package_tag_id=package_tag_id, error=_coerce_solution_api_error(exc), retried=True, ) retry_items = retry_result.get("items") if isinstance(retry_result.get("items"), list) else [] if retry_items: items = retry_items source_shape = retry_result.get("source_shape") retried = True exact = [ item for item in items if isinstance(item, dict) and str(item.get("tagName") or "").strip().casefold() == normalized_name.casefold() ] fuzzy_candidates = [ item for item in items if isinstance(item, dict) and normalized_name.casefold() in str(item.get("tagName") or "").strip().casefold() ] if len(exact) == 1: match = exact[0] return { "status": "resolved", "matched_via": "exact", "tag_id": match.get("tagId"), "tag_name": match.get("tagName"), "candidates": [], "source_shape": source_shape, "retried": retried, } candidates = [ {"tag_id": item.get("tagId"), "tag_name": item.get("tagName")} for item in (exact or fuzzy_candidates)[:10] if isinstance(item, dict) ] detail = f"package_name '{normalized_name}' did not resolve uniquely by exact name" error_code = "AMBIGUOUS_PACKAGE" if not exact and not fuzzy_candidates: detail = f"package_name '{normalized_name}' was not found" error_code = "PACKAGE_NOT_FOUND" error_fields = _solution_error_fields( category="config", detail=detail, suggested_next_call=None, stage="app", ) return { "status": "failed", "response": { "status": "failed", "mode": "plan", "stage": "app", "errors": [{"category": "config", "detail": detail}], "error_code": error_code, "recoverable": True, "missing_required_fields": error_fields["missing_required_fields"], "unknown_fields": error_fields["unknown_fields"], "invalid_field_types": error_fields["invalid_field_types"], "package_resolution": { "status": "failed", "requested_name": normalized_name, "resolution_policy": "exact_name_only", "candidates": candidates, "source_shape": source_shape, "retried": retried, }, "suggested_next_call": { "tool_name": "solution_build_app_from_requirements", "arguments": {"mode": "plan", "title": "", "requirement_text": "", "package_name": normalized_name}, }, }, } def _resolve_builder_existing_app_reference( self, *, profile: str, app_key: str, package_resolution: dict[str, Any], update_only: bool, ) -> dict[str, Any]: """执行内部辅助逻辑。""" normalized_app_key = str(app_key or "").strip() if not normalized_app_key: if update_only: detail = "update target requires app_key" error_fields = _solution_error_fields(category="config", detail=detail, suggested_next_call=None, stage="app") error_fields["missing_required_fields"] = sorted( dict.fromkeys(list(error_fields.get("missing_required_fields", [])) + ["app_key"]) ) error_fields["suggested_schema_call"] = None return { "status": "failed", "response": { "status": "failed", "stage": "app", "errors": [{"category": "config", "detail": detail}], **error_fields, }, } return {"status": "none"} app_tools = AppTools(self.sessions, self.backend) try: base_info = app_tools.app_get_base(profile=profile, app_key=normalized_app_key, include_raw=True).get("result") or {} except Exception as exc: # noqa: BLE001 detail = f"failed to resolve existing app '{normalized_app_key}': {exc}" error_fields = _solution_error_fields(category="config", detail=detail, suggested_next_call=None, stage="app") error_fields["suggested_schema_call"] = None return { "status": "failed", "response": { "status": "failed", "stage": "app", "errors": [{"category": "config", "detail": detail}], **error_fields, }, } raw_tag_ids = base_info.get("tagIds") if isinstance(base_info.get("tagIds"), list) else [] tag_ids = [tag_id for tag_id in (_coerce_count(item) for item in raw_tag_ids) if tag_id is not None] effective_package_resolution = deepcopy(package_resolution) if package_resolution.get("status") == "resolved": resolved_tag_id = _coerce_count(package_resolution.get("tag_id")) if resolved_tag_id is not None and tag_ids and resolved_tag_id not in tag_ids: detail = ( f"app_key '{normalized_app_key}' does not belong to requested package " f"'{package_resolution.get('tag_name') or resolved_tag_id}'" ) error_fields = _solution_error_fields(category="config", detail=detail, suggested_next_call=None, stage="app") error_fields["error_code"] = "APP_PACKAGE_MISMATCH" error_fields["suggested_schema_call"] = None return { "status": "failed", "response": { "status": "failed", "stage": "app", "errors": [{"category": "config", "detail": detail}], "existing_app": { "app_key": normalized_app_key, "app_name": base_info.get("formTitle"), "tag_ids": tag_ids, }, **error_fields, }, } elif package_resolution.get("status") == "new_package": if len(tag_ids) == 1: package_tools = PackageTools(self.sessions, self.backend) package_detail = package_tools.package_get(profile=profile, tag_id=tag_ids[0], include_raw=False).get("result") or {} effective_package_resolution = { "status": "resolved", "matched_via": "existing_app", "tag_id": tag_ids[0], "tag_name": package_detail.get("tagName"), "candidates": [], } elif len(tag_ids) > 1: detail = f"app_key '{normalized_app_key}' belongs to multiple packages; provide package_tag_id or package_name explicitly" error_fields = _solution_error_fields(category="config", detail=detail, suggested_next_call=None, stage="app") error_fields["error_code"] = "AMBIGUOUS_PACKAGE" error_fields["suggested_schema_call"] = None return { "status": "failed", "response": { "status": "failed", "stage": "app", "errors": [{"category": "config", "detail": detail}], "existing_app": { "app_key": normalized_app_key, "app_name": base_info.get("formTitle"), "tag_ids": tag_ids, }, **error_fields, }, } return { "status": "resolved", "app_key": normalized_app_key, "app_name": base_info.get("formTitle"), "package_resolution": effective_package_resolution, "tag_ids": tag_ids, } def _requirements_failure_response( self, *, mode: str, build_id: str, title: str, requirement_text: str, package_resolution: dict[str, Any], app_key: str, update_only: bool, layout_style: str, publish: bool, run_label: str | None, detail: str, error_code: str | None = None, recoverable: bool | None = None, missing_required_fields: list[str] | None = None, invalid_field_types: list[str] | None = None, error_details: dict[str, Any] | None = None, ) -> dict[str, Any]: """执行内部辅助逻辑。""" suggested_next_call = ( {"tool_name": "solution_build_app", "arguments": {"mode": "plan", "build_id": build_id}} if build_id and self._load_generated_app_spec(build_id) else _requirements_next_call( next_mode="plan" if mode in {"", "preflight"} else mode, build_id=build_id or None, title=title, requirement_text=requirement_text, package_resolution=package_resolution, app_key=app_key, update_only=update_only, layout_style=layout_style, publish=publish, run_label=run_label, ) ) error_fields = _solution_error_fields( category="config", detail=detail, suggested_next_call=suggested_next_call, stage="app", ) if error_code: error_fields["error_code"] = error_code if recoverable is not None: error_fields["recoverable"] = recoverable if missing_required_fields: error_fields["missing_required_fields"] = sorted( dict.fromkeys(list(error_fields.get("missing_required_fields", [])) + list(missing_required_fields)) ) if invalid_field_types: error_fields["invalid_field_types"] = sorted( dict.fromkeys(list(error_fields.get("invalid_field_types", [])) + list(invalid_field_types)) ) error_fields["suggested_schema_call"] = None return { "status": "failed", "stage": "app", "mode": mode, "build_id": build_id, "errors": [{"category": "config", "detail": detail}], "package_resolution": package_resolution, "next_strategy": "switch_to_explicit_app_stage" if build_id and self._load_generated_app_spec(build_id) else "refine_requirements", "failure_signature": None, "stage_failure_count": 0, "details": deepcopy(error_details or {}), "suggested_requirement_hint": (error_details or {}).get("suggested_requirement_hint") if isinstance(error_details, dict) else None, "suggested_patch": deepcopy((error_details or {}).get("suggested_patch")) if isinstance(error_details, dict) else None, **error_fields, } def _requirements_success_response( self, *, mode: str, title: str, requirement_text: str, package_resolution: dict[str, Any], layout_style: str, publish: bool, run_label: str | None, generated: GeneratedAppBuild, stage_result: dict[str, Any], include_generated_spec: bool, ) -> dict[str, Any]: """执行内部辅助逻辑。""" next_call = stage_result.get("suggested_next_call") response = { "status": stage_result.get("status"), "stage": "app", "mode": mode, "build_id": stage_result.get("build_id"), "build_path": stage_result.get("build_path"), "run_path": stage_result.get("run_path"), "artifacts": stage_result.get("artifacts", {}), "errors": stage_result.get("errors", []), "package_resolution": package_resolution, "generated_app_summary": generated.summary, "resolved_layout_style": generated.summary.get("resolved_layout_style"), "field_selection_policy": generated.summary.get("field_selection_policy"), "excluded_advanced_fields": generated.summary.get("excluded_advanced_fields"), "execution_step_count": len((stage_result.get("execution_plan") or {}).get("steps", [])), "suggested_next_call": next_call, "views_strategy": _resolve_app_stage_views_strategy(stage_result), } if isinstance(stage_result.get("verification"), dict): verification = stage_result["verification"] response["verification"] = verification if "package_attached" in verification: response["package_attached"] = verification.get("package_attached") if isinstance(verification.get("apps"), list): response["tag_ids_after"] = { str(item.get("entity_id")): item.get("tag_ids_after") for item in verification["apps"] if isinstance(item, dict) and item.get("entity_id") is not None } if isinstance(verification.get("views_created"), list): response["views_created"] = verification.get("views_created") for key in ( "error_code", "recoverable", "missing_required_fields", "unknown_fields", "invalid_field_types", "invalid_enum_values", "next_strategy", "failure_signature", "stage_failure_count", ): if key in stage_result: response[key] = stage_result[key] if include_generated_spec: response["generated_app_spec"] = generated.app_spec return response def deep_merge(base: dict[str, Any], patch: dict[str, Any]) -> dict[str, Any]: merged = deepcopy(base) for key, value in patch.items(): if isinstance(value, dict) and isinstance(merged.get(key), dict): merged[key] = deep_merge(merged[key], value) else: merged[key] = deepcopy(value) return merged def _extract_publish_results(step_results: dict[str, Any]) -> dict[str, Any]: return { step_name: step_result for step_name, step_result in step_results.items() if step_name.startswith("publish.") } def _sanitize_json_value(value: Any) -> Any: if value is None or isinstance(value, (str, int, float, bool)): return value if isinstance(value, BaseModel): return _sanitize_json_value(value.model_dump(mode="json")) if isinstance(value, dict): return {str(key): _sanitize_json_value(item) for key, item in value.items()} if isinstance(value, (list, tuple, set)): return [_sanitize_json_value(item) for item in value] if isinstance(value, Exception): return str(value) if hasattr(value, "__fspath__"): return str(value) return str(value) def _sanitize_errors(errors: list[dict[str, Any]]) -> list[dict[str, Any]]: sanitized: list[dict[str, Any]] = [] for error in errors: payload = _sanitize_json_value(error) if isinstance(payload, dict): sanitized.append(payload) continue sanitized.append({"category": "runtime", "detail": payload}) return sanitized def _generate_build_id(*, run_label: str | None, stage_name: str) -> str: prefix = _slug_fragment(run_label) or stage_name return f"{prefix}-{uuid4().hex[:10]}" def _generate_session_id(*, metadata: dict[str, Any], stage: str | None) -> str: base = _slug_fragment(str(metadata.get("goal") or metadata.get("task") or metadata.get("run_label") or "")) or stage or "design" return f"{base}-{uuid4().hex[:10]}" def _slug_fragment(value: str | None) -> str: if not value: return "" text = "".join(ch.lower() if ch.isalnum() else "-" for ch in value).strip("-") compact = "-".join(part for part in text.split("-") if part) return compact[:32].rstrip("-") def _solution_build_next_call(*, tool_name: str, mode: str, build_id: str | None) -> dict[str, Any] | None: if not build_id: return None if mode not in STAGED_BUILD_MODES: return None if mode == "preflight": next_mode = "plan" elif mode == "plan": next_mode = "apply" else: next_mode = "repair" return {"tool_name": tool_name, "arguments": {"mode": next_mode, "build_id": build_id}} def _normalize_staged_build_mode(mode: str) -> str: normalized = str(mode or "").strip().lower() if not normalized: return normalized return STAGED_BUILD_MODE_ALIASES.get(normalized, normalized) def _requirements_next_call( *, next_mode: str, build_id: str | None, title: str, requirement_text: str, package_resolution: dict[str, Any], app_key: str, update_only: bool, layout_style: str, publish: bool, run_label: str | None, ) -> dict[str, Any]: arguments: dict[str, Any] = { "mode": next_mode, "title": title, "requirement_text": requirement_text, "layout_style": layout_style, "publish": publish, } if build_id: arguments["build_id"] = build_id if run_label: arguments["run_label"] = run_label if isinstance(package_resolution.get("tag_id"), int) and package_resolution["tag_id"] > 0: arguments["package_tag_id"] = package_resolution["tag_id"] elif isinstance(package_resolution.get("tag_name"), str) and package_resolution["tag_name"]: arguments["package_name"] = package_resolution["tag_name"] if isinstance(app_key, str) and app_key: arguments["app_key"] = app_key if update_only or arguments.get("app_key"): arguments["update_only"] = True return {"tool_name": "solution_build_app_from_requirements", "arguments": arguments} def _rewrite_requirements_next_call( *, result_next_call: dict[str, Any] | None, title: str, requirement_text: str, package_resolution: dict[str, Any], app_key: str, update_only: bool, layout_style: str, publish: bool, run_label: str | None, ) -> dict[str, Any] | None: if not isinstance(result_next_call, dict): return None arguments = result_next_call.get("arguments") if not isinstance(arguments, dict): return None next_mode = arguments.get("mode") if not isinstance(next_mode, str) or not next_mode: return None build_id = arguments.get("build_id") return _requirements_next_call( next_mode=next_mode, build_id=build_id if isinstance(build_id, str) else None, title=title, requirement_text=requirement_text, package_resolution=package_resolution, app_key=app_key, update_only=update_only, layout_style=layout_style, publish=publish, run_label=run_label, ) def _requirements_target( package_resolution: dict[str, Any], *, app_key: str = "", update_only: bool = False, ) -> dict[str, Any]: tag_id = package_resolution.get("tag_id") target: dict[str, Any] = {} if isinstance(tag_id, int) and tag_id > 0: target["package_tag_id"] = tag_id if isinstance(app_key, str) and app_key: target["app_key"] = app_key if update_only or target.get("app_key"): target["mode"] = "update" target["update_only"] = True return target def _merge_app_target( target: dict[str, Any] | None, *, app_key: str, package_tag_id: int, update_only: bool, ) -> dict[str, Any]: merged = deepcopy(target) if isinstance(target, dict) else {} if isinstance(app_key, str) and app_key.strip(): merged["app_key"] = app_key.strip() if isinstance(package_tag_id, int) and package_tag_id > 0: merged["package_tag_id"] = package_tag_id if update_only or merged.get("app_key"): merged["mode"] = "update" merged["update_only"] = True return merged def _target_is_update_only(target: dict[str, Any]) -> bool: if not target: return False if target.get("update_only") is True: return True mode = target.get("mode") return isinstance(mode, str) and mode.strip().lower() == "update" def _resolve_target_app_keys(*, target: dict[str, Any], entity_ids: list[str]) -> dict[str, str]: if not target: return {} resolved: dict[str, str] = {} direct_app_key = target.get("app_key") if isinstance(direct_app_key, str) and direct_app_key.strip(): if len(entity_ids) != 1: raise ValueError("target.app_key requires a single-entity app stage; use target.app_keys for multiple entities") resolved[entity_ids[0]] = direct_app_key.strip() raw_mapping = target.get("app_keys") if isinstance(raw_mapping, dict): for entity_id, app_key in raw_mapping.items(): if not isinstance(entity_id, str) or entity_id not in entity_ids: raise ValueError(f"target.app_keys contains unknown entity '{entity_id}'") if not isinstance(app_key, str) or not app_key.strip(): raise ValueError(f"target.app_keys['{entity_id}'] must be a non-empty string") resolved[entity_id] = app_key.strip() return resolved def _seed_existing_app_artifacts(*, assembly: BuildAssemblyStore, existing_app_targets: dict[str, str]) -> None: if not existing_app_targets: return artifacts = assembly.get_artifacts() apps_artifact = artifacts.get("apps", {}) if isinstance(artifacts.get("apps"), dict) else {} for entity_id, app_key in existing_app_targets.items(): existing = apps_artifact.get(entity_id, {}) if isinstance(apps_artifact.get(entity_id), dict) else {} next_artifact = deepcopy(existing) next_artifact["app_key"] = app_key next_artifact["reused"] = True next_artifact["target_mode"] = "update" apps_artifact[entity_id] = next_artifact artifacts["apps"] = apps_artifact assembly.set_artifacts(artifacts) def _design_session_next_call(*, action: str, session_id: str | None) -> dict[str, Any] | None: if action == "start" and not session_id: return {"tool_name": "solution_design_session", "arguments": {"action": "start"}} if session_id: return {"tool_name": "solution_design_session", "arguments": {"action": "get", "session_id": session_id}} return None def _design_summary_next_call(summary: dict[str, Any]) -> dict[str, Any] | None: session_id = summary.get("session_id") if not session_id: return None if summary.get("status") == "finalized": return None next_stage = summary.get("next_stage") if next_stage == "finalize": return {"tool_name": "solution_design_session", "arguments": {"action": "finalize", "session_id": session_id}} if next_stage: return { "tool_name": "solution_design_session", "arguments": {"action": "submit_stage", "session_id": session_id, "stage": next_stage}, } return {"tool_name": "solution_design_session", "arguments": {"action": "get", "session_id": session_id}} def _normalize_solution_schema_stage(stage: str) -> str | None: normalized = str(stage or "").strip() if not normalized: return None return SOLUTION_SCHEMA_STAGE_ALIASES.get(normalized.lower()) or SOLUTION_SCHEMA_STAGE_ALIASES.get(normalized) def _normalize_solution_schema_intent(intent: str) -> str | None: normalized = str(intent or "").strip() if not normalized: return None return SOLUTION_SCHEMA_INTENT_ALIASES.get(normalized.lower()) or SOLUTION_SCHEMA_INTENT_ALIASES.get(normalized) def _solution_schema_example_payload( *, stage: str, intent: str, include_examples: bool = False, include_selected_example: bool = False, ) -> dict[str, Any]: tool_name, payload_key = _solution_schema_tool_metadata(stage) examples = _solution_schema_examples(stage) selected = deepcopy(examples.get(intent) or examples["minimal"]) call_example = _solution_schema_call_example(stage) response = { "status": "ok", "stage": stage, "intent": intent, "tool_name": tool_name, "payload_key": payload_key, "available_intents": sorted(examples.keys()), "required_fields": _solution_schema_required_fields(stage), "optional_fields": _solution_schema_optional_fields(stage), "common_errors": _solution_schema_common_errors(stage), "notes": _solution_schema_notes(stage), "call_example": call_example, "suggested_next_call": deepcopy(call_example), } if include_selected_example: response["selected_example"] = selected if include_examples: response["examples"] = examples return response def _solution_schema_tool_metadata(stage: str) -> tuple[str, str]: mapping = { "app": ("solution_build_app", "app_spec"), "app_update": ("solution_build_app", "app_spec"), "flow": ("solution_build_flow", "flow_spec"), "views": ("solution_build_views", "views_spec"), "analytics_portal": ("solution_build_analytics_portal", "analytics_portal_spec"), "navigation": ("solution_build_navigation", "navigation_spec"), "app_flow": ("solution_build_app_flow", "app_flow_spec"), "all": ("solution_build_all", "solution_spec"), } return mapping[stage] def _solution_schema_call_example(stage: str) -> dict[str, Any]: tool_name, _payload_key = _solution_schema_tool_metadata(stage) arguments: dict[str, Any] = {"mode": "preflight"} if stage == "app_update": arguments.update({"app_key": "APP_123456", "update_only": True}) return {"tool_name": tool_name, "arguments": arguments} def _solution_schema_required_fields(stage: str) -> list[str]: fields = { "app": ["solution_name", "entities[].entity_id", "entities[].display_name", "entities[].kind", "entities[].fields[]"], "app_update": ["solution_name", "entities[].entity_id", "entities[].display_name", "entities[].kind", "entities[].fields[]", "app_key"], "flow": ["entities[].entity_id", "entities[].workflow or entities[].lifecycle_stages"], "views": ["entities[].entity_id", "entities[].views[]"], "analytics_portal": ["entities[].entity_id or portal.sections[]"], "navigation": ["navigation.items[]"], "app_flow": ["solution_name", "entities[]"], "all": ["solution_name", "entities[]"], } return fields[stage] def _solution_schema_optional_fields(stage: str) -> list[str]: fields = { "app": ["summary", "business_context", "package", "roles", "publish_policy", "preferences", "entities[].form_layout", "entities[].sample_records"], "app_update": ["package_tag_id", "publish_policy", "preferences", "entities[].form_layout", "entities[].sample_records", "update_only"], "flow": ["solution_name", "roles", "publish_policy", "entities[].status_field_id", "entities[].owner_field_id", "entities[].start_field_id", "entities[].end_field_id"], "views": ["solution_name", "metadata"], "analytics_portal": ["solution_name", "publish_policy", "portal.name", "portal.sections[]"], "navigation": ["solution_name", "publish_policy", "navigation.enabled"], "app_flow": ["summary", "package", "roles", "publish_policy", "preferences"], "all": ["portal", "navigation", "roles", "publish_policy", "preferences", "business_context"], } return fields[stage] def _solution_schema_common_errors(stage: str) -> list[str]: errors = { "app": [ "Use enum values exactly, for example field.type='text' or kind='transaction'.", "Each field_id must be unique within an entity.", "Do not include workflow/views/charts in app_spec. Use solution_build_flow or later stage tools.", ], "app_update": [ "app_key is required for update mode and must point to an existing app.", "Updating an app uses the same app_spec structure, but the tool target must stay in update mode.", "Do not omit entity_id; updates still need a deterministic entity identity.", ], "flow": [ "The target entity must already exist in the build manifest or from a prior solution_build_app run.", "status_field_id must point to an existing field, usually 'status'.", "Do not include fields or form_layout in flow_spec.", ], "views": [ "View field_ids must reference existing fields on the entity.", "Run solution_build_app first so the entity exists.", ], "analytics_portal": [ "Chart field references must already exist on the entity.", "Portal sections that reference views require solution_build_views to run first.", ], "navigation": [ "Navigation must include a package target item as the single published root.", ], "app_flow": [ "If app and flow are authored separately, prefer solution_build_app then solution_build_flow.", ], "all": [ "Use full solution_spec only when the agent can provide all stages coherently.", ], } return errors[stage] def _solution_schema_notes(stage: str) -> list[str]: notes = { "app": [ "Use solution_build_app for package, entity, field, layout, and sample record design.", "If you are unsure, call solution_schema_example(stage='app', intent='minimal') before generating payloads.", ], "app_update": [ "Use solution_build_app with app_key and update_only=true when modifying an existing app.", "Update mode is not allowed to fall back to new package or new app creation.", ], "flow": [ "Use solution_build_flow only after app/schema exists.", "workflow.nodes uses node_type enum values like start, audit, fill, copy, branch, condition.", ], "views": ["views_spec is purely for view definitions, not fields or workflow."], "analytics_portal": ["analytics_portal_spec may include charts, portal, or both."], "navigation": ["navigation_spec only manages top-level published navigation."], "app_flow": ["solution_build_app_flow remains available for compatibility, but app + flow split is preferred."], "all": ["solution_build_all is best for coherent multi-stage system builds."], } return notes[stage] def _solution_schema_examples(stage: str) -> dict[str, dict[str, Any]]: examples = { "app": { "minimal": { "solution_name": "Demo Workspace", "entities": [ { "entity_id": "demo_request", "display_name": "测试申请", "kind": "transaction", "title_field_id": "title", "fields": [ {"field_id": "title", "label": "标题", "type": "text", "required": True}, ], "form_layout": {"rows": [{"field_ids": ["title"]}]}, } ], }, "full": { "solution_name": "Demo Workspace", "summary": "通用测试应用", "package": {"name": "测试包"}, "entities": [ { "entity_id": "demo_request", "display_name": "测试申请", "kind": "transaction", "title_field_id": "title", "status_field_id": "status", "fields": [ {"field_id": "title", "label": "标题", "type": "text", "required": True}, {"field_id": "status", "label": "状态", "type": "single_select", "options": ["待处理", "处理中", "已完成"]}, {"field_id": "owner", "label": "负责人", "type": "member"}, {"field_id": "amount", "label": "金额", "type": "amount"}, ], "form_layout": { "sections": [ {"section_id": "basic", "title": "基础信息", "rows": [{"field_ids": ["title", "status"]}, {"field_ids": ["owner", "amount"]}]} ] }, "sample_records": [{"record_id": "demo-1", "values": {"title": "测试记录", "status": "待处理", "amount": 1000}}], } ], }, "demo": { "solution_name": "全字段演示", "package": {"name": "字段演示"}, "entities": [ { "entity_id": "field_demo", "display_name": "字段演示", "kind": "transaction", "title_field_id": "title", "fields": [ {"field_id": "title", "label": "标题", "type": "text", "required": True}, {"field_id": "description", "label": "说明", "type": "long_text"}, {"field_id": "amount", "label": "金额", "type": "amount"}, {"field_id": "due_at", "label": "截止时间", "type": "datetime"}, {"field_id": "assignee", "label": "处理人", "type": "member"}, {"field_id": "team", "label": "所属部门", "type": "department"}, {"field_id": "category", "label": "类型", "type": "single_select", "options": ["A", "B", "C"]}, {"field_id": "tags", "label": "标签", "type": "multi_select", "options": ["高优先级", "新建", "内部"]}, {"field_id": "phone", "label": "手机号", "type": "phone"}, {"field_id": "email", "label": "邮箱", "type": "email"}, {"field_id": "attachment", "label": "附件", "type": "attachment"}, {"field_id": "enabled", "label": "启用", "type": "boolean"}, ], "form_layout": { "sections": [ {"section_id": "basic", "title": "基础信息", "rows": [{"field_ids": ["title", "category"]}, {"field_ids": ["description"]}]}, {"section_id": "coordination", "title": "协同", "rows": [{"field_ids": ["assignee", "team"]}, {"field_ids": ["due_at", "amount"]}]}, {"section_id": "contact", "title": "联系与附件", "rows": [{"field_ids": ["phone", "email"]}, {"field_ids": ["attachment", "enabled"]}]}, ] }, } ], }, }, "app_update": { "minimal": { "solution_name": "Demo Workspace", "entities": [ { "entity_id": "demo_request", "display_name": "测试申请", "kind": "transaction", "title_field_id": "title", "fields": [ {"field_id": "title", "label": "工单标题", "type": "text", "required": True}, ], "form_layout": {"rows": [{"field_ids": ["title"]}]}, } ], }, "full": { "solution_name": "Demo Workspace", "entities": [ { "entity_id": "demo_request", "display_name": "测试申请", "kind": "transaction", "title_field_id": "title", "status_field_id": "status", "fields": [ {"field_id": "title", "label": "工单标题", "type": "text", "required": True}, {"field_id": "status", "label": "状态", "type": "single_select", "options": ["待处理", "处理中", "已完成"]}, {"field_id": "priority", "label": "优先级", "type": "single_select", "options": ["P0", "P1", "P2"]}, ], "form_layout": { "sections": [ {"section_id": "basic", "title": "基础信息", "rows": [{"field_ids": ["title", "status"]}, {"field_ids": ["priority"]}]} ] }, } ], }, "demo": { "solution_name": "Demo Workspace", "entities": [ { "entity_id": "customer_order", "display_name": "客户订单", "kind": "transaction", "title_field_id": "title", "fields": [ {"field_id": "title", "label": "订单标题", "type": "text", "required": True}, {"field_id": "customer_name", "label": "客户名称", "type": "text"}, {"field_id": "amount", "label": "订单金额", "type": "amount"}, ], "form_layout": {"rows": [{"field_ids": ["title", "customer_name"]}, {"field_ids": ["amount"]}]}, } ], }, }, "flow": { "minimal": { "solution_name": "Demo Workspace", "entities": [ { "entity_id": "demo_request", "status_field_id": "status", "workflow": { "enabled": True, "nodes": [ {"node_id": "approve", "name": "审批", "node_type": "audit"}, ], }, } ], }, "full": { "solution_name": "Demo Workspace", "roles": [{"role_id": "approver", "name": "审批人"}], "entities": [ { "entity_id": "demo_request", "status_field_id": "status", "owner_field_id": "owner", "workflow": { "enabled": True, "nodes": [ {"node_id": "start", "name": "发起", "node_type": "start"}, {"node_id": "approve", "name": "直属审批", "node_type": "audit", "assignees": {"roles": ["approver"]}}, ], }, } ], }, "demo": { "solution_name": "Demo Workspace", "entities": [ { "entity_id": "field_demo", "status_field_id": "status", "workflow": { "enabled": True, "nodes": [ {"node_id": "start", "name": "发起", "node_type": "start"}, {"node_id": "manager_review", "name": "经理审批", "node_type": "audit"}, {"node_id": "copy_notice", "name": "结果抄送", "node_type": "copy"}, ], }, } ], }, }, "views": { "minimal": { "solution_name": "Demo Workspace", "entities": [ {"entity_id": "demo_request", "views": [{"view_id": "table", "name": "表格", "type": "table", "field_ids": ["title"], "being_default": True}]} ], }, "full": { "solution_name": "Demo Workspace", "entities": [ { "entity_id": "demo_request", "views": [ {"view_id": "table", "name": "表格", "type": "table", "field_ids": ["title", "status", "owner"], "being_default": True}, {"view_id": "board", "name": "看板", "type": "board", "field_ids": ["title", "status"], "group_by_field_id": "status"}, ], } ], }, "demo": { "solution_name": "Demo Workspace", "entities": [ { "entity_id": "field_demo", "views": [ {"view_id": "table", "name": "字段表格", "type": "table", "field_ids": ["title", "category", "assignee"], "being_default": True}, {"view_id": "card", "name": "卡片", "type": "card", "field_ids": ["title", "category", "due_at"]}, ], } ], }, }, "analytics_portal": { "minimal": { "solution_name": "Demo Workspace", "entities": [{"entity_id": "demo_request", "charts": [{"chart_id": "total", "name": "总量", "chart_type": "target", "indicator_field_ids": ["title"], "config": {"aggregate": "count"}}]}], }, "full": { "solution_name": "Demo Workspace", "entities": [{"entity_id": "demo_request", "charts": [{"chart_id": "total", "name": "总量", "chart_type": "target", "indicator_field_ids": ["title"], "config": {"aggregate": "count"}}]}], "portal": {"name": "Demo 首页", "sections": [{"section_id": "total", "title": "总量", "source_type": "chart", "entity_id": "demo_request", "chart_id": "total"}]}, }, "demo": { "solution_name": "Demo Workspace", "entities": [{"entity_id": "field_demo", "charts": [{"chart_id": "by_category", "name": "按类型统计", "chart_type": "pie", "dimension_field_ids": ["category"], "config": {"aggregate": "count"}}]}], "portal": {"name": "字段演示首页", "sections": [{"section_id": "category_chart", "title": "类型分布", "source_type": "chart", "entity_id": "field_demo", "chart_id": "by_category"}]}, }, }, "navigation": { "minimal": {"solution_name": "Demo Workspace", "navigation": {"items": [{"item_id": "package_root", "title": "Demo Workspace", "target_type": "package"}]}}, "full": {"solution_name": "Demo Workspace", "navigation": {"items": [{"item_id": "package_root", "title": "Demo Workspace", "target_type": "package"}, {"item_id": "demo_request", "title": "测试申请", "target_type": "app", "entity_id": "demo_request"}]}}, "demo": {"solution_name": "Demo Workspace", "navigation": {"items": [{"item_id": "package_root", "title": "字段演示", "target_type": "package"}, {"item_id": "field_demo", "title": "字段演示", "target_type": "app", "entity_id": "field_demo"}]}}, }, "app_flow": { "minimal": { "solution_name": "Demo Workspace", "entities": [ { "entity_id": "demo_request", "display_name": "测试申请", "kind": "transaction", "title_field_id": "title", "status_field_id": "status", "fields": [ {"field_id": "title", "label": "标题", "type": "text", "required": True}, {"field_id": "status", "label": "状态", "type": "single_select", "options": ["待处理", "已完成"]}, ], "form_layout": {"rows": [{"field_ids": ["title", "status"]}]}, "workflow": {"enabled": True, "nodes": [{"node_id": "approve", "name": "审批", "node_type": "audit"}]}, } ], }, "full": {}, "demo": {}, }, "all": { "minimal": { "solution_name": "Demo Workspace", "entities": [ { "entity_id": "demo_request", "display_name": "测试申请", "kind": "transaction", "title_field_id": "title", "fields": [{"field_id": "title", "label": "标题", "type": "text", "required": True}], "form_layout": {"rows": [{"field_ids": ["title"]}]}, } ], }, "full": { "solution_name": "Demo Workspace", "package": {"name": "Demo Package"}, "entities": [ { "entity_id": "demo_request", "display_name": "测试申请", "kind": "transaction", "title_field_id": "title", "status_field_id": "status", "fields": [ {"field_id": "title", "label": "标题", "type": "text", "required": True}, {"field_id": "status", "label": "状态", "type": "single_select", "options": ["待处理", "处理中", "已完成"]}, ], "form_layout": {"rows": [{"field_ids": ["title", "status"]}]}, "workflow": {"enabled": True, "nodes": [{"node_id": "approve", "name": "审批", "node_type": "audit"}]}, "views": [{"view_id": "table", "name": "表格", "type": "table", "field_ids": ["title", "status"], "being_default": True}], "charts": [{"chart_id": "total", "name": "总量", "chart_type": "target", "indicator_field_ids": ["title"], "config": {"aggregate": "count"}}], } ], "portal": {"name": "Demo 首页", "sections": [{"section_id": "total", "title": "总量", "source_type": "chart", "entity_id": "demo_request", "chart_id": "total"}]}, "navigation": {"items": [{"item_id": "package_root", "title": "Demo Workspace", "target_type": "package"}]}, }, "demo": {}, }, } if not examples["app_flow"]["full"]: examples["app_flow"]["full"] = deepcopy(examples["all"]["full"]) if not examples["app_flow"]["demo"]: examples["app_flow"]["demo"] = deepcopy(examples["app"]["demo"]) examples["app_flow"]["demo"]["entities"][0]["status_field_id"] = "status" examples["app_flow"]["demo"]["entities"][0]["fields"].append( {"field_id": "status", "label": "状态", "type": "single_select", "options": ["待处理", "处理中", "已完成"]} ) examples["app_flow"]["demo"]["entities"][0]["workflow"] = { "enabled": True, "nodes": [ {"node_id": "start", "name": "发起", "node_type": "start"}, {"node_id": "approve", "name": "审批", "node_type": "audit"}, ], } if not examples["all"]["demo"]: examples["all"]["demo"] = deepcopy(examples["all"]["full"]) return examples[stage] def _solution_error_fields( *, category: str, detail: Any, suggested_next_call: dict[str, Any] | None, stage: str | None = None, ) -> dict[str, Any]: detail_text = str(detail) missing_required_fields: list[str] = [] unknown_fields: list[str] = [] invalid_field_types: list[str] = [] invalid_enum_values: list[str] = [] if isinstance(detail, list): for item in detail: if not isinstance(item, dict): continue loc = ".".join(str(part) for part in item.get("loc", [])) err_type = str(item.get("type", "")) if err_type.endswith("missing") and loc: missing_required_fields.append(loc) if err_type == "extra_forbidden" and loc: unknown_fields.append(loc) if (err_type.endswith("_type") or err_type in {"enum", "literal_error", "model_type", "list_type", "dict_type"}) and loc: invalid_field_types.append(loc) if err_type in {"enum", "literal_error"} and loc: invalid_enum_values.append(loc) if "build_id is required" in detail_text: missing_required_fields.append("build_id") if "session_id is required" in detail_text: missing_required_fields.append("session_id") if "solution_spec must be a non-empty object" in detail_text: missing_required_fields.append("solution_spec") if "design_patch must be a non-empty object" in detail_text: missing_required_fields.append("design_patch") recoverable = category in {"config", "validation"} error_code = f"SOLUTION_{category.upper()}" normalized_stage = "app" if stage == "app_flow" else stage suggested_schema_call = None if recoverable and normalized_stage in SOLUTION_SCHEMA_STAGES: suggested_schema_call = { "tool_name": "solution_schema_example", "arguments": {"stage": normalized_stage, "intent": "minimal"}, } return { "error_code": error_code, "recoverable": recoverable, "missing_required_fields": sorted(dict.fromkeys(missing_required_fields)), "unknown_fields": sorted(dict.fromkeys(unknown_fields)), "invalid_field_types": sorted(dict.fromkeys(invalid_field_types)), "invalid_enum_values": sorted(dict.fromkeys(invalid_enum_values)), "suggested_next_call": suggested_next_call, "suggested_schema_call": suggested_schema_call, } def _builder_package_resolution_failed( *, package_name: str, package_tag_id: int, error: QingflowApiError, retried: bool, ) -> dict[str, Any]: permission_restricted_listing = ( package_tag_id <= 0 and not is_auth_like_error(error) and backend_code_int(error) in {40002, 40027} ) if package_tag_id > 0: detail = f"failed to resolve package_tag_id '{package_tag_id}': {error.message}" elif permission_restricted_listing: detail = ( f"failed to resolve package '{package_name}' because package listing is permission-restricted; " "provide package_tag_id explicitly or use an account that can list packages" ) else: detail = f"failed to resolve package '{package_name}': {error.message}" category = "config" if permission_restricted_listing else str(error.category or "backend") error_fields = _solution_error_fields(category=category, detail=detail, suggested_next_call=None, stage="app") error_fields["error_code"] = "AUTH_REQUIRED" if is_auth_like_error(error) else "PACKAGE_RESOLVE_FAILED" return { "status": "failed", "response": { "status": "failed", "mode": "plan", "stage": "app", "errors": [{"category": category, "detail": detail}], "package_resolution": { "status": "failed", "requested_name": package_name or None, "requested_tag_id": package_tag_id or None, "resolution_policy": "exact_name_only" if package_name else "tag_id_only", "candidates": [], "source_shape": None, "retried": retried, "transport_error": { "http_status": error.http_status, "backend_code": error.backend_code, "category": error.category, "request_id": error.request_id, }, }, **error_fields, }, } def _coerce_solution_api_error(error: Exception) -> QingflowApiError: if isinstance(error, QingflowApiError): return error if isinstance(error, RuntimeError): try: payload = json.loads(str(error)) except json.JSONDecodeError: payload = None if isinstance(payload, dict) and payload.get("category") and payload.get("message"): details = payload.get("details") return QingflowApiError( category=str(payload.get("category")), message=str(payload.get("message")), backend_code=payload.get("backend_code"), request_id=payload.get("request_id"), http_status=payload.get("http_status"), details=details if isinstance(details, dict) else None, ) return QingflowApiError(category="runtime", message=str(error)) def _coerce_count(value: Any) -> int | None: if isinstance(value, bool) or value is None: return None if isinstance(value, int): return value if isinstance(value, float): return int(value) if isinstance(value, str): text = value.strip() if not text: return None try: return int(text) except ValueError: return None return None def _failure_signature_for_stage(*, stage_name: str, stage_payload: dict[str, Any], errors: list[dict[str, Any]]) -> str | None: error_steps = sorted({str(error.get("step_name")) for error in errors if isinstance(error, dict) and error.get("step_name")}) if stage_name != "app": return fingerprint_payload( { "stage": stage_name, "error_steps": error_steps, } ) entities = stage_payload.get("entities") if not isinstance(entities, list) or not entities: return fingerprint_payload( { "stage": stage_name, "error_steps": error_steps, } ) normalized_entities = [_normalized_entity_failure_payload(entity) for entity in entities if isinstance(entity, dict)] if not normalized_entities: return fingerprint_payload( { "stage": stage_name, "error_steps": error_steps, } ) return fingerprint_payload( { "stage": stage_name, "entities": normalized_entities, "error_steps": error_steps, } ) def _normalized_entity_failure_payload(entity: dict[str, Any]) -> dict[str, Any]: fields = entity.get("fields") layout = entity.get("form_layout") normalized_fields = [] if isinstance(fields, list): for field in fields: if not isinstance(field, dict): continue normalized_fields.append( { "field_id": field.get("field_id"), "type": field.get("type"), "required": bool(field.get("required")), } ) return { "entity_id": entity.get("entity_id"), "fields": normalized_fields, "layout": _normalized_layout_signature(layout if isinstance(layout, dict) else {}), } def _normalized_layout_signature(layout: dict[str, Any]) -> dict[str, Any]: rows = layout.get("rows") sections = layout.get("sections") normalized_rows: list[list[str]] = [] if isinstance(rows, list): for row in rows: if isinstance(row, dict): normalized_rows.append(list(row.get("field_ids", []))) normalized_sections = [] if isinstance(sections, list): for section in sections: if not isinstance(section, dict): continue normalized_sections.append( { "section_id": section.get("section_id"), "rows": [ list(row.get("field_ids", [])) for row in section.get("rows", []) if isinstance(row, dict) ], } ) return {"rows": normalized_rows, "sections": normalized_sections} def _looks_like_full_solution_spec(payload: dict[str, Any]) -> bool: if not isinstance(payload, dict): return False if not isinstance(payload.get("entities"), list): return False return any( key in payload for key in ( "portal", "navigation", "requirements", "success_metrics", "roles", "package", "business_context", ) ) def _project_stage_payload(stage_name: str, solution_spec: dict[str, Any]) -> dict[str, Any]: if stage_name == "app": entities = deepcopy(solution_spec.get("entities", [])) for entity in entities: if not isinstance(entity, dict): continue entity.pop("workflow", None) entity.pop("lifecycle_stages", None) entity.pop("views", None) entity.pop("charts", None) projected = { "solution_name": solution_spec.get("solution_name"), "summary": solution_spec.get("summary"), "business_context": deepcopy(solution_spec.get("business_context", {})), "package": deepcopy(solution_spec.get("package", {})), "entities": entities, "roles": deepcopy(solution_spec.get("roles", [])), "publish_policy": deepcopy(solution_spec.get("publish_policy", {})), "preferences": deepcopy(solution_spec.get("preferences", {})), "assumptions": deepcopy(solution_spec.get("assumptions", [])), "constraints": deepcopy(solution_spec.get("constraints", [])), "metadata": deepcopy(solution_spec.get("metadata", {})), } return {key: value for key, value in projected.items() if value is not None} if stage_name == "flow": projected_entities: list[dict[str, Any]] = [] for entity in solution_spec.get("entities", []): if not isinstance(entity, dict): continue flow_entity = {"entity_id": entity.get("entity_id")} for key in ("workflow", "lifecycle_stages", "status_field_id", "owner_field_id", "start_field_id", "end_field_id", "metadata"): if key in entity: flow_entity[key] = deepcopy(entity[key]) projected_entities.append(flow_entity) return { "solution_name": solution_spec.get("solution_name"), "entities": projected_entities, "roles": deepcopy(solution_spec.get("roles", [])), "publish_policy": deepcopy(solution_spec.get("publish_policy", {})), "metadata": deepcopy(solution_spec.get("metadata", {})), } if stage_name == "app_flow": entities = deepcopy(solution_spec.get("entities", [])) for entity in entities: if not isinstance(entity, dict): continue entity.pop("views", None) entity.pop("charts", None) projected = { "solution_name": solution_spec.get("solution_name"), "summary": solution_spec.get("summary"), "business_context": deepcopy(solution_spec.get("business_context", {})), "package": deepcopy(solution_spec.get("package", {})), "entities": entities, "roles": deepcopy(solution_spec.get("roles", [])), "publish_policy": deepcopy(solution_spec.get("publish_policy", {})), "preferences": deepcopy(solution_spec.get("preferences", {})), "assumptions": deepcopy(solution_spec.get("assumptions", [])), "constraints": deepcopy(solution_spec.get("constraints", [])), "metadata": deepcopy(solution_spec.get("metadata", {})), } return {key: value for key, value in projected.items() if value is not None} if stage_name == "views": return { "solution_name": solution_spec.get("solution_name"), "entities": [ { "entity_id": entity.get("entity_id"), "views": deepcopy(entity.get("views", [])), } for entity in solution_spec.get("entities", []) if isinstance(entity, dict) and entity.get("views") ], "metadata": deepcopy(solution_spec.get("metadata", {})), } if stage_name == "analytics_portal": return { "solution_name": solution_spec.get("solution_name"), "entities": [ { "entity_id": entity.get("entity_id"), "charts": deepcopy(entity.get("charts", [])), } for entity in solution_spec.get("entities", []) if isinstance(entity, dict) and entity.get("charts") ], "portal": deepcopy(solution_spec.get("portal", {})), "publish_policy": deepcopy(solution_spec.get("publish_policy", {})), "metadata": deepcopy(solution_spec.get("metadata", {})), } if stage_name == "navigation": return { "solution_name": solution_spec.get("solution_name"), "navigation": deepcopy(solution_spec.get("navigation", {})), "publish_policy": deepcopy(solution_spec.get("publish_policy", {})), "metadata": deepcopy(solution_spec.get("metadata", {})), } raise ValueError(f"unsupported stage '{stage_name}'") def _split_solution_spec_into_stage_payloads(solution_spec: dict[str, Any]) -> dict[str, dict[str, Any]]: return {stage_name: _project_stage_payload(stage_name, solution_spec) for stage_name in STAGE_ORDER} def _stage_payload_is_empty(stage_name: str, stage_payload: dict[str, Any]) -> bool: if stage_name == "app_flow": return not bool(stage_payload.get("entities")) if stage_name == "views": return not any( isinstance(entity, dict) and entity.get("views") for entity in stage_payload.get("entities", []) ) if stage_name == "analytics_portal": has_charts = any( isinstance(entity, dict) and entity.get("charts") for entity in stage_payload.get("entities", []) ) portal_payload = stage_payload.get("portal", {}) if isinstance(stage_payload.get("portal"), dict) else {} has_portal = bool(portal_payload.get("sections")) and portal_payload.get("enabled", True) is not False return not has_charts and not has_portal if stage_name == "navigation": navigation_payload = stage_payload.get("navigation", {}) if isinstance(stage_payload.get("navigation"), dict) else {} return not (navigation_payload.get("enabled", True) and navigation_payload.get("items")) raise ValueError(f"unsupported stage '{stage_name}'") def _stage_skip_reason(stage_name: str) -> str: reasons = { "app_flow": "No app flow entities were provided for this build stage.", "views": "No view definitions were provided, so the views stage was skipped.", "analytics_portal": "No charts or enabled portal sections were provided, so the analytics stage was skipped.", "navigation": "No enabled navigation items were provided, so the navigation stage was skipped.", } return reasons.get(stage_name, "Stage skipped because no applicable payload was provided.") def _verification_error_payload(exc: Exception) -> dict[str, Any]: payload: dict[str, Any] = {"detail": str(exc)} if isinstance(exc, QingflowApiError): payload["transport_error"] = { "http_status": exc.http_status, "backend_code": exc.backend_code, "category": exc.category, "request_id": exc.request_id, } payload["request_id"] = exc.request_id payload["backend_code"] = exc.backend_code payload["http_status"] = exc.http_status return payload def _append_verification_error(verification: dict[str, Any], scope: str, exc: Exception) -> None: errors = verification.setdefault("errors", []) if isinstance(errors, list): error_payload = _verification_error_payload(exc) error_payload["scope"] = scope error_payload["message"] = str(exc) errors.append(error_payload) PREFERRED_RECORD_TITLE_LABELS = ( "名称", "客户名称", "医疗机构名称", "联系人姓名", "商机名称", "销售机会名称", "合同名称", "采购合同名称", "拜访主题", "主题", "项目名称", "任务名称", ) def _readback_record_page( record_tools: RecordTools, profile: str, app_key: str, *, entity_spec: dict[str, Any] | None = None, field_meta: dict[str, Any] | None = None, ) -> dict[str, Any]: select_columns = _readback_select_columns(entity_spec=entity_spec, field_meta=field_meta) fallback: dict[str, Any] = {} for list_type in (8, 9): response = record_tools.record_query( profile=profile, query_mode="list", app_key=app_key, apply_id=None, page_num=1, page_size=20, requested_pages=1, scan_max_pages=1, query_key=None, filters=[], sorts=[], max_rows=20, max_columns=max(1, len(select_columns)), select_columns=select_columns, amount_column=None, time_range={}, stat_policy={}, strict_full=False, output_profile="compact", list_type=list_type, view_key=None, view_name=None, ) data = response.get("data") if isinstance(response, dict) else None list_payload = data.get("list") if isinstance(data, dict) and isinstance(data.get("list"), dict) else {} pagination = list_payload.get("pagination") if isinstance(list_payload.get("pagination"), dict) else {} rows = list_payload.get("rows") if isinstance(list_payload.get("rows"), list) else [] merged_page = { "list": rows, "backend_reported_total": _coerce_count(pagination.get("backend_reported_total")), "returned_rows": _coerce_count(pagination.get("returned_items")) or len(rows), "effective_count": _coerce_count(pagination.get("result_amount")), "list_type": list_type, "list_type_label": get_record_list_type_label(list_type), "page_num": _coerce_count(pagination.get("page_num")) or 1, "page_size": _coerce_count(pagination.get("page_size")) or 20, } if not fallback and isinstance(merged_page, dict): fallback = merged_page if rows: return merged_page return fallback def _extract_record_titles( rows: list[Any], *, entity_spec: dict[str, Any] | None = None, field_meta: dict[str, Any] | None = None, ) -> list[str]: preferred_titles = set(PREFERRED_RECORD_TITLE_LABELS) title_labels, title_que_ids = _title_field_hints(entity_spec=entity_spec, field_meta=field_meta) titles: list[str] = [] for row in rows: if not isinstance(row, dict): continue answers = row.get("answers") selected: str | None = None fallback: str | None = None if isinstance(answers, list): for answer in answers: if not isinstance(answer, dict): continue normalized_values = _extract_answer_values(answer) if not normalized_values: continue candidate = " / ".join(normalized_values) answer_title = answer.get("queTitle") or answer.get("title") answer_que_id = _coerce_count(answer.get("queId")) if answer_que_id in title_que_ids or answer_title in title_labels: selected = candidate break if fallback is None and not _looks_like_secondary_answer(answer_title, candidate): fallback = candidate if answer_title in preferred_titles: selected = candidate break title = selected or fallback if title is None: for label in [*sorted(title_labels), *[item for item in PREFERRED_RECORD_TITLE_LABELS if item not in title_labels]]: candidate = _normalize_row_title(row.get(label)) if candidate and not _looks_like_secondary_answer(label, candidate): title = candidate break if title is None: for key, value in row.items(): if key in {"apply_id", "id"}: continue candidate = _normalize_row_title(value) if candidate and not _looks_like_secondary_answer(key, candidate): title = candidate break if title is None: for key in ("title", "name", "recordName", "applyName"): value = row.get(key) candidate = _normalize_row_title(value) if candidate: title = candidate break if title: titles.append(title) return titles[:5] def _readback_select_columns( *, entity_spec: dict[str, Any] | None, field_meta: dict[str, Any] | None, ) -> list[str]: title_labels, _ = _title_field_hints(entity_spec=entity_spec, field_meta=field_meta) by_label = field_meta.get("by_label", {}) if isinstance(field_meta, dict) and isinstance(field_meta.get("by_label"), dict) else {} selected: list[str] = [] def append(label: Any) -> None: if not isinstance(label, str) or not label or label in selected: return if by_label and label not in by_label: return selected.append(label) for label in sorted(title_labels): append(label) for label in PREFERRED_RECORD_TITLE_LABELS: append(label) if isinstance(by_label, dict): for label in by_label: append(label) if len(selected) >= 5: break if not selected and isinstance(entity_spec, dict): for field in entity_spec.get("fields", []): if not isinstance(field, dict): continue append(field.get("label")) if len(selected) >= 5: break return selected[:5] or ["名称"] def _title_field_hints( *, entity_spec: dict[str, Any] | None, field_meta: dict[str, Any] | None, ) -> tuple[set[str], set[int]]: title_labels: set[str] = set() title_que_ids: set[int] = set() if not isinstance(entity_spec, dict): return title_labels, title_que_ids title_field_candidates = {str(entity_spec.get("title_field_id") or "title")} for field in entity_spec.get("fields", []): if not isinstance(field, dict): continue field_id = field.get("field_id") field_label = field.get("label") field_config = field.get("config", {}) if isinstance(field.get("config"), dict) else {} if field_id in title_field_candidates or field_config.get("role") == "title": if isinstance(field_label, str) and field_label: title_labels.add(field_label) if isinstance(field_id, str): title_field_candidates.add(field_id) if isinstance(field_meta, dict): by_field_id = field_meta.get("by_field_id", {}) if isinstance(by_field_id, dict): for field_id in title_field_candidates: que_id = _coerce_count(by_field_id.get(field_id)) if que_id is not None: title_que_ids.add(que_id) return title_labels, title_que_ids def _extract_answer_values(answer: dict[str, Any]) -> list[str]: values = answer.get("values") normalized_values: list[str] = [] if isinstance(values, list): for value in values: if isinstance(value, str) and value: normalized_values.append(value) continue if isinstance(value, dict): for key in ("value", "name", "text", "label"): item = value.get(key) if isinstance(item, str) and item: normalized_values.append(item) break elif isinstance(values, str) and values: normalized_values.append(values) return normalized_values def _normalize_row_title(value: Any) -> str | None: if isinstance(value, str): text = value.strip() return text or None if isinstance(value, (int, float)) and not isinstance(value, bool): return str(value) if isinstance(value, list): parts = [_normalize_row_title(item) for item in value] normalized = [part for part in parts if isinstance(part, str) and part] return " / ".join(normalized) if normalized else None if isinstance(value, dict): for key in ("value", "name", "text", "label"): candidate = _normalize_row_title(value.get(key)) if candidate: return candidate return None def _looks_like_secondary_answer(answer_title: Any, candidate: str) -> bool: if not isinstance(candidate, str) or not candidate: return True title = str(answer_title or "") secondary_markers = ("状态", "日期", "时间", "负责人", "处理人", "优先级", "金额", "费用", "电话", "类型") if any(marker in title for marker in secondary_markers): return True if len(candidate) >= 10 and candidate[4] == "-" and candidate[7] == "-": return True return False def _count_portal_components(payload: dict[str, Any]) -> int | None: if not isinstance(payload, dict): return None for key in ("components", "sections", "dashItemList"): items = payload.get(key) if isinstance(items, list): return len(items) return None def _extract_navigation_items(items: list[Any]) -> list[dict[str, Any]]: extracted: list[dict[str, Any]] = [] for item in items: if not isinstance(item, dict): continue result = item.get("result") if isinstance(item.get("result"), dict) else {} result_payload = result.get("result") if isinstance(result.get("result"), dict) else {} summarized = { "item_id": item.get("item_id"), "navigation_item_id": result_payload.get("navigationItemId") or result_payload.get("id"), "title": result_payload.get("navigationItemName") or result_payload.get("title"), } extracted.append({key: value for key, value in summarized.items() if value is not None}) return extracted def _resolve_app_stage_views_strategy(stage_result: dict[str, Any]) -> str: verification = stage_result.get("verification") if isinstance(verification, dict): strategy = verification.get("views_strategy") if isinstance(strategy, str) and strategy: return strategy return "not_included" def _resolve_target_package_tag_id(target: dict[str, Any]) -> int | None: if not target: return None package_tag_id = target.get("package_tag_id") if package_tag_id is None and isinstance(target.get("package"), dict): package_tag_id = target["package"].get("tag_id") if package_tag_id is None: return None if not isinstance(package_tag_id, int) or package_tag_id <= 0: raise ValueError("target.package_tag_id must be a positive integer") return package_tag_id def _bind_existing_package_target(compiled: CompiledSolution, package_tag_id: int) -> CompiledSolution: compiled.package_payload = None compiled.execution_plan = build_execution_plan(compiled.normalized_spec, include_package=False, attach_package=True) for entity in compiled.entities: entity.app_create_payload["tagIds"] = [package_tag_id] return compiled def _bind_existing_app_target(compiled: CompiledSolution) -> CompiledSolution: compiled.package_payload = None compiled.execution_plan = build_execution_plan(compiled.normalized_spec, include_package=False, attach_package=False) return compiled def _merge_stage_manifest(*, stage_name: str, manifest: dict[str, Any], stage_payload: dict[str, Any]) -> dict[str, Any]: merged = deepcopy(manifest or default_manifest()) if stage_name == "app_flow": return _merge_app_flow_manifest(merged, stage_payload) if stage_name == "views": return _merge_views_manifest(merged, stage_payload) if stage_name == "analytics_portal": return _merge_analytics_portal_manifest(merged, stage_payload) if stage_name == "navigation": return _merge_navigation_manifest(merged, stage_payload) raise ValueError(f"unsupported stage '{stage_name}'") def _merge_app_flow_manifest(manifest: dict[str, Any], stage_payload: dict[str, Any]) -> dict[str, Any]: entity_map = {entity["entity_id"]: deepcopy(entity) for entity in manifest.get("entities", [])} merged_entities: list[dict[str, Any]] = [] staged_entity_ids: set[str] = set() for entity in stage_payload.get("entities", []): existing = entity_map.get(entity["entity_id"], {}) next_entity = deepcopy(existing) next_entity.update(deepcopy(entity)) next_entity["views"] = deepcopy(existing.get("views", [])) next_entity["charts"] = deepcopy(existing.get("charts", [])) merged_entities.append(next_entity) staged_entity_ids.add(entity["entity_id"]) for entity_id, entity in entity_map.items(): if entity_id not in staged_entity_ids: merged_entities.append(deepcopy(entity)) manifest["solution_name"] = stage_payload.get("solution_name", manifest.get("solution_name")) manifest["summary"] = stage_payload.get("summary", manifest.get("summary")) manifest["business_context"] = deepcopy(stage_payload.get("business_context", manifest.get("business_context", {}))) manifest["package"] = deepcopy(_merge_dict(manifest.get("package", {}), stage_payload.get("package", {}))) manifest["entities"] = merged_entities if "roles" in stage_payload: manifest["roles"] = deepcopy(stage_payload["roles"]) manifest["publish_policy"] = deepcopy(_merge_dict(manifest.get("publish_policy", {}), stage_payload.get("publish_policy", {}))) manifest["preferences"] = deepcopy(_merge_dict(manifest.get("preferences", {}), stage_payload.get("preferences", {}))) if "assumptions" in stage_payload: manifest["assumptions"] = deepcopy(stage_payload["assumptions"]) if "constraints" in stage_payload: manifest["constraints"] = deepcopy(stage_payload["constraints"]) manifest["metadata"] = deepcopy(_merge_dict(manifest.get("metadata", {}), stage_payload.get("metadata", {}))) return manifest def _merge_views_manifest(manifest: dict[str, Any], stage_payload: dict[str, Any]) -> dict[str, Any]: entity_map = {entity["entity_id"]: deepcopy(entity) for entity in manifest.get("entities", [])} for entity_patch in stage_payload.get("entities", []): entity_id = entity_patch["entity_id"] if entity_id not in entity_map: raise ValueError(f"views stage references unknown entity '{entity_id}'") entity_map[entity_id]["views"] = deepcopy(entity_patch.get("views", [])) manifest["entities"] = list(entity_map.values()) manifest["metadata"] = deepcopy(_merge_dict(manifest.get("metadata", {}), stage_payload.get("metadata", {}))) return manifest def _merge_analytics_portal_manifest(manifest: dict[str, Any], stage_payload: dict[str, Any]) -> dict[str, Any]: entity_map = {entity["entity_id"]: deepcopy(entity) for entity in manifest.get("entities", [])} for entity_patch in stage_payload.get("entities", []): entity_id = entity_patch["entity_id"] if entity_id not in entity_map: raise ValueError(f"analytics stage references unknown entity '{entity_id}'") entity_map[entity_id]["charts"] = deepcopy(entity_patch.get("charts", [])) manifest["entities"] = list(entity_map.values()) if "portal" in stage_payload: portal_payload = deepcopy(stage_payload["portal"]) if portal_payload.get("sections"): manifest["portal"] = deepcopy(_merge_dict(manifest.get("portal", {}), portal_payload)) manifest["portal"]["enabled"] = portal_payload.get("enabled", True) manifest.setdefault("preferences", {}) manifest["preferences"]["create_portal"] = bool(portal_payload.get("enabled", True)) elif portal_payload.get("enabled") is False: manifest["portal"] = deepcopy(_merge_dict(manifest.get("portal", {}), portal_payload)) manifest.setdefault("preferences", {}) manifest["preferences"]["create_portal"] = False if "publish_policy" in stage_payload: manifest["publish_policy"] = deepcopy(_merge_dict(manifest.get("publish_policy", {}), stage_payload["publish_policy"])) manifest["metadata"] = deepcopy(_merge_dict(manifest.get("metadata", {}), stage_payload.get("metadata", {}))) return manifest def _merge_navigation_manifest(manifest: dict[str, Any], stage_payload: dict[str, Any]) -> dict[str, Any]: navigation_payload = deepcopy(stage_payload.get("navigation", {})) manifest["navigation"] = deepcopy(_merge_dict(manifest.get("navigation", {}), navigation_payload)) if navigation_payload.get("items"): manifest["navigation"]["enabled"] = navigation_payload.get("enabled", True) manifest.setdefault("preferences", {}) manifest["preferences"]["create_navigation"] = bool(navigation_payload.get("enabled", True)) if "publish_policy" in stage_payload: manifest["publish_policy"] = deepcopy(_merge_dict(manifest.get("publish_policy", {}), stage_payload["publish_policy"])) manifest["metadata"] = deepcopy(_merge_dict(manifest.get("metadata", {}), stage_payload.get("metadata", {}))) return manifest def _filter_compiled_solution( compiled: CompiledSolution, *, stage_name: str, stage_payload: dict[str, Any], stage_variant: str | None = None, ) -> CompiledSolution: included_step_names = _stage_step_names( compiled.execution_plan, stage_name=stage_name, stage_payload=stage_payload, stage_variant=stage_variant, ) allowed = set(included_step_names) filtered_steps = [ ExecutionStep( step_name=step.step_name, resource_type=step.resource_type, resource_ref=step.resource_ref, description=step.description, depends_on=[dependency for dependency in step.depends_on if dependency in allowed], ) for step in compiled.execution_plan.steps if step.step_name in allowed ] filtered = deepcopy(compiled) filtered.execution_plan = ExecutionPlan(steps=filtered_steps) return filtered def _stage_step_names( plan: ExecutionPlan, *, stage_name: str, stage_payload: dict[str, Any], stage_variant: str | None = None, ) -> list[str]: variant = stage_variant or stage_name if variant == "navigation": return [step.step_name for step in plan.steps if step.step_name in {"navigation.create", "publish.navigation"}] entity_ids = { entity["entity_id"] for entity in stage_payload.get("entities", []) if isinstance(entity, dict) and isinstance(entity.get("entity_id"), str) } included: list[str] = [] for step in plan.steps: if variant == "app": if ( step.step_name == "package.create" or step.step_name.startswith("role.create.") or any( step.step_name.startswith(prefix) for prefix in ( "app.create.", "package.attach.", "form.base.", "form.relations.", "publish.form.", "publish.app.", "seed_data.", ) ) ): if "." in step.step_name and step.resource_ref not in entity_ids and not step.step_name.startswith("role.create.") and step.step_name != "package.create": continue included.append(step.step_name) continue if variant == "flow": if ( step.step_name.startswith("role.create.") or step.step_name.startswith("workflow.") or step.step_name.startswith("publish.workflow.") or step.step_name.startswith("publish.app.") ): if "." in step.step_name and step.resource_ref not in entity_ids and not step.step_name.startswith("role.create."): continue included.append(step.step_name) continue if variant == "app_flow": if ( step.step_name == "package.create" or step.step_name.startswith("role.create.") or any( step.step_name.startswith(prefix) for prefix in ( "app.create.", "package.attach.", "form.base.", "form.relations.", "workflow.", "publish.form.", "publish.workflow.", "publish.app.", "seed_data.", ) ) ): if "." in step.step_name and step.resource_ref not in entity_ids and not step.step_name.startswith("role.create.") and step.step_name != "package.create": continue included.append(step.step_name) continue if variant == "views": if step.resource_ref in entity_ids and (step.step_name.startswith("views.") or step.step_name.startswith("publish.app.")): included.append(step.step_name) continue if variant == "analytics_portal": if step.resource_ref in entity_ids and (step.step_name.startswith("charts.") or step.step_name.startswith("publish.app.")): included.append(step.step_name) continue if step.step_name in {"portal.create", "publish.portal"}: included.append(step.step_name) continue return included def _merge_dict(base: dict[str, Any], patch: dict[str, Any]) -> dict[str, Any]: merged = deepcopy(base) for key, value in patch.items(): merged[key] = deepcopy(value) return merged def _stage_run_id(build_id: str, stage_name: str) -> str: return f"{build_id}--{stage_name}" def _build_status_for_stage(stage_name: str, result_status: str) -> str: if result_status != "success": return result_status return "success" if stage_name == "navigation" else "ready" def _build_summary(assembly: BuildAssemblyStore) -> dict[str, Any]: summary = assembly.summary() stage_specs = summary.get("stage_specs", {}) if isinstance(summary.get("stage_specs"), dict) else {} stage_history = summary.get("stage_history", []) if isinstance(summary.get("stage_history"), list) else [] latest_entries: dict[str, dict[str, Any]] = {} for entry in stage_history: stage = entry.get("stage") if isinstance(stage, str): latest_entries[stage] = entry stage_statuses: dict[str, str] = {} for stage in STAGE_ORDER: latest = latest_entries.get(stage) if latest and latest.get("status"): stage_statuses[stage] = str(latest["status"]) continue stage_statuses[stage] = "draft" if stage_specs.get(stage) else "pending" ready_stages = [stage for stage in STAGE_ORDER if not _missing_stage_prerequisites(stage_name=stage, assembly=assembly, stage_payload=stage_specs.get(stage, {}) or {})] completed_stages = [stage for stage, status in stage_statuses.items() if status in {"success", "ready", "skipped"}] next_stage = next((stage for stage in STAGE_ORDER if stage_statuses.get(stage) not in {"success", "ready", "skipped"}), None) prerequisites = { stage: _missing_stage_prerequisites(stage_name=stage, assembly=assembly, stage_payload=stage_specs.get(stage, {}) or {}) for stage in STAGE_ORDER } return { "build_id": summary.get("build_id"), "status": summary.get("status"), "build_path": summary.get("build_path"), "stage_statuses": stage_statuses, "completed_stages": completed_stages, "ready_stages": ready_stages, "next_recommended_stage": next_stage, "missing_prerequisites": prerequisites, "stage_history": stage_history, } def _missing_stage_prerequisites(*, stage_name: str, assembly: BuildAssemblyStore, stage_payload: dict[str, Any]) -> list[str]: manifest = assembly.get_manifest() stage_specs = (assembly.data.get("stage_specs", {}) or {}) if isinstance(assembly.data.get("stage_specs"), dict) else {} entity_map = { entity.get("entity_id"): entity for entity in manifest.get("entities", []) if isinstance(entity, dict) and isinstance(entity.get("entity_id"), str) } missing: list[str] = [] if stage_name in {"app_flow", "app"}: return missing if stage_name == "flow": if not entity_map and not stage_specs.get("app_flow"): return ["Run solution_build_app first to define package, apps, and fields before configuring workflow."] for entity in stage_payload.get("entities", []): entity_id = entity.get("entity_id") if entity_id not in entity_map: missing.append(f"Entity '{entity_id}' is not defined yet. Run solution_build_app first.") return list(dict.fromkeys(missing)) if not entity_map and not stage_specs.get("app_flow"): return ["Run solution_build_app_flow first to define package, apps, and form/workflow entities."] if stage_name == "views": for entity in stage_payload.get("entities", []): entity_id = entity.get("entity_id") if entity_id not in entity_map: missing.append(f"Entity '{entity_id}' is not defined yet. Run solution_build_app_flow first.") return missing if stage_name == "analytics_portal": for entity in stage_payload.get("entities", []): entity_id = entity.get("entity_id") if entity_id not in entity_map: missing.append(f"Entity '{entity_id}' is not defined yet. Run solution_build_app_flow first.") portal_payload = stage_payload.get("portal", {}) if isinstance(stage_payload.get("portal"), dict) else {} for section in portal_payload.get("sections", []): if section.get("source_type") == "view": entity_id = section.get("entity_id") view_id = section.get("view_id") entity = entity_map.get(entity_id) if isinstance(entity_id, str) else None known_view_ids = { view.get("view_id") for view in (entity.get("views", []) if isinstance(entity, dict) else []) if isinstance(view, dict) } if view_id not in known_view_ids: missing.append( f"Portal section '{section.get('section_id')}' references view '{view_id}' on entity '{entity_id}'. Run solution_build_views first." ) return list(dict.fromkeys(missing)) if stage_name == "navigation": navigation_payload = stage_payload.get("navigation", {}) if isinstance(stage_payload.get("navigation"), dict) else {} package_targets = [ item for item in navigation_payload.get("items", []) if isinstance(item, dict) and item.get("target_type") == "package" ] if not package_targets: missing.append("Navigation stage must include a package target item as the single published navigation root.") return list(dict.fromkeys(missing)) return missing