from __future__ import annotations from copy import deepcopy 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 ..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.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 .workflow_tools import WorkflowTools from .workspace_tools import WorkspaceTools STAGED_BUILD_MODES = {"preflight", "plan", "apply", "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", "flow", "views", "analytics_portal", "navigation", "app_flow", "all"} SOLUTION_SCHEMA_INTENTS = {"minimal", "full", "demo"} class SolutionTools(ToolBase): def register(self, mcp: FastMCP) -> None: @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, 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 {}, publish=publish, run_label=run_label, target=target or {}, repair_patch=repair_patch or {}, ) @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", ) -> dict[str, Any]: return self.solution_schema_example(stage=stage, intent=intent) @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]: 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()) 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]: 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], publish: bool, run_label: str | None, target: dict[str, Any], repair_patch: dict[str, Any], ) -> dict[str, Any]: return self._stage_build( profile=profile, mode=mode, build_id=build_id, stage_name="app_flow", stage_payload=app_spec, stage_model=AppBuildSpec, publish=publish, run_label=run_label, target=target, repair_patch=repair_patch, public_stage_name="app", tool_name="solution_build_app", projection_name="app", ) 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]: 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]: 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()) 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]: 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) -> dict[str, Any]: if stage not in SOLUTION_SCHEMA_STAGES: return { "stage": stage, "intent": intent, "status": "failed", "errors": [{"category": "config", "detail": f"stage must be one of: {', '.join(sorted(SOLUTION_SCHEMA_STAGES))}"}], **_solution_error_fields( category="config", detail=f"stage must be one of: {', '.join(sorted(SOLUTION_SCHEMA_STAGES))}", suggested_next_call=None, ), } if intent not in SOLUTION_SCHEMA_INTENTS: return { "stage": stage, "intent": intent, "status": "failed", "errors": [{"category": "config", "detail": f"intent must be one of: {', '.join(sorted(SOLUTION_SCHEMA_INTENTS))}"}], **_solution_error_fields( category="config", detail=f"intent must be one of: {', '.join(sorted(SOLUTION_SCHEMA_INTENTS))}", suggested_next_call=None, ), } return _solution_schema_example_payload(stage=stage, intent=intent) 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]: 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]: 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]: 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()) 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) 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) 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 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"]) assembly.add_stage_history( { "stage": stage_name, "mode": mode, "status": result["status"], "run_path": result["run_path"], "execution_step_count": len(filtered_compiled.execution_plan.steps), } ) assembly.mark_status(_build_status_for_stage(stage_name, result["status"])) return 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=result["status"], build_summary=_build_summary(assembly), tool_name=tool_name, ) 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), workflow_tools=WorkflowTools(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).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 _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, ) 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, ) -> dict[str, Any]: return { "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, "build_summary": build_summary, "suggested_next_call": _solution_build_next_call( tool_name=tool_name, mode=mode, build_id=build_id, ), } 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 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) -> str: 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 _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 _solution_schema_example_payload(*, stage: str, intent: str) -> 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"]) return { "status": "ok", "stage": stage, "intent": intent, "tool_name": tool_name, "payload_key": payload_key, "required_fields": _solution_schema_required_fields(stage), "optional_fields": _solution_schema_optional_fields(stage), "selected_example": selected, "examples": examples, "common_errors": _solution_schema_common_errors(stage), "notes": _solution_schema_notes(stage), "suggested_next_call": { "tool_name": tool_name, "arguments": {"mode": "preflight", payload_key: selected}, }, } def _solution_schema_tool_metadata(stage: str) -> tuple[str, str]: mapping = { "app": ("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_required_fields(stage: str) -> list[str]: fields = { "app": ["solution_name", "entities[].entity_id", "entities[].display_name", "entities[].kind", "entities[].fields[]"], "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"], "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.", ], "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.", ], "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"]}]}, ] }, } ], }, }, "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] = [] 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 "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)), "suggested_next_call": suggested_next_call, "suggested_schema_call": suggested_schema_call, } 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 _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 _append_verification_error(verification: dict[str, Any], scope: str, exc: Exception) -> None: errors = verification.setdefault("errors", []) if isinstance(errors, list): errors.append( { "scope": scope, "message": str(exc), } ) 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_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) for entity in compiled.entities: entity.app_create_payload["tagIds"] = [package_tag_id] 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.", "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.", "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