from __future__ import annotations from concurrent.futures import ThreadPoolExecutor, as_completed from copy import deepcopy import json import time from typing import Any from pydantic import ValidationError from ..builder_facade.button_style_catalog import ( BUTTON_BACKGROUND_COLORS, BUTTON_ICONS, BUTTON_STYLE_PRESETS, BUTTON_TEXT_COLORS, ) from ..public_surface import public_builder_contract_tool_names from ..config import DEFAULT_PROFILE from ..errors import QingflowApiError, backend_code_int, backend_code_value_int from ..json_types import JSONObject from ..builder_facade.models import ( AssociatedResourcesApplyRequest, ChartApplyRequest, CustomButtonsApplyRequest, CustomButtonPatch, FIELD_TYPE_ALIASES, FIELD_TYPE_ID_ALIASES, FieldPatch, FieldRemovePatch, FieldUpdatePatch, FlowPreset, FlowNodePatch, FlowPlanRequest, FlowTransitionPatch, LayoutApplyMode, LayoutPlanRequest, LayoutPreset, LayoutSectionPatch, PortalApplyRequest, PublicButtonPlacement, PublicButtonTriggerAction, PublicFieldType, PublicRelationMode, PublicChartType, PublicViewType, SchemaPlanRequest, VisibilityPatch, ViewFilterOperator, ViewPartialPatch, ViewUpsertPatch, ViewsPreset, ViewsPlanRequest, public_view_partial_payload, public_view_upsert_payload, ) from ..builder_facade.service import AiBuilderFacade, INTEGRATION_OUTPUT_TARGET_FIELD_TYPES from ..solution.compiler.icon_utils import ( GENERIC_WORKSPACE_ICON_NAMES, WORKSPACE_ICON_COLORS, WORKSPACE_ICON_NAMES, normalize_workspace_icon_name, validate_workspace_icon_choice, workspace_icon_catalog_payload, workspace_icon_config, ) from .app_tools import AppTools from .base import ToolBase from .custom_button_tools import CustomButtonTools from .directory_tools import DirectoryTools from .package_tools import PackageTools from .portal_tools import PortalTools from .qingbi_report_tools import QingbiReportTools from .role_tools import RoleTools from .solution_tools import SolutionTools from .view_tools import ViewTools from .workflow_tools import WorkflowTools def _normalize_builder_view_key(value: str) -> str: raw = str(value or "").strip() if raw.startswith("custom:"): return raw.split(":", 1)[1].strip() return raw def _payload_get(payload: JSONObject, *keys: str, default: Any = None) -> Any: for key in keys: if key in payload: return payload.get(key) return default def _payload_list(payload: JSONObject, *keys: str, default: list | None = None) -> list: value = _payload_get(payload, *keys, default=default if default is not None else []) return value if isinstance(value, list) else [] def _payload_bool(payload: JSONObject, *keys: str, default: bool = True) -> bool: value = _payload_get(payload, *keys, default=default) if isinstance(value, bool): return value if isinstance(value, str): normalized = value.strip().lower() if normalized in {"1", "true", "yes", "y", "on"}: return True if normalized in {"0", "false", "no", "n", "off"}: return False return bool(value) def _elapsed_ms(started_at: float) -> int: return max(0, int((time.perf_counter() - started_at) * 1000)) def _merge_duration_breakdown(payload: JSONObject, **items: int | None) -> None: breakdown = payload.get("duration_breakdown_ms") if not isinstance(breakdown, dict): breakdown = {} payload["duration_breakdown_ms"] = breakdown for key, value in items.items(): if value is None: continue breakdown[key] = int(value) def _inline_form_layout_was_handled(payload: JSONObject) -> bool: inline_layout = payload.get("inline_form_layout") return isinstance(inline_layout, dict) and bool(inline_layout.get("requested")) PUBLIC_STABLE_FLOW_NODE_TYPES = ["start", "approve", "fill", "copy", "webhook", "end"] BUILDER_APPLY_SCHEMA_VERSION = "builder.apply.v1" BUILDER_APPLY_TOOL_NAMES = { "package_apply", "app_schema_apply", "app_layout_apply", "app_flow_apply", "app_views_apply", "app_custom_buttons_apply", "app_associated_resources_apply", "app_charts_apply", "portal_apply", "portal_delete", "app_publish_verify", } MULTI_APP_SCHEMA_APPLY_PARALLEL_LIMIT = 10 VIEW_APPLY_PARALLEL_LIMIT = 20 PORTAL_INLINE_CHART_MAX_WORKERS = 10 class AiBuilderTools(ToolBase): """AI Builder 工具（中文名：AI 搭建编排）。 类型：应用搭建编排工具。 主要职责： 1. 编排 schema/layout/views/flow/charts 的计划与应用； 2. 聚合 builder 侧读写与校验能力； 3. 将复杂搭建流程收敛为可执行步骤。 """ def __init__(self, sessions, backend) -> None: """执行内部辅助逻辑。""" super().__init__(sessions, backend) self._facade = AiBuilderFacade( apps=AppTools(sessions, backend), buttons=CustomButtonTools(sessions, backend), packages=PackageTools(sessions, backend), views=ViewTools(sessions, backend), workflows=WorkflowTools(sessions, backend), portals=PortalTools(sessions, backend), charts=QingbiReportTools(sessions, backend), roles=RoleTools(sessions, backend), directory=DirectoryTools(sessions, backend), solutions=SolutionTools(sessions, backend), ) def _apply_app_batch(self, *, tool_name: str, profile: str, apps: list[JSONObject], apply_one) -> JSONObject: started_at = time.perf_counter() batch_item_apply_ms = 0 child_total_ms = 0 def finish(payload: JSONObject) -> JSONObject: _merge_duration_breakdown( payload, batch_item_apply_ms=batch_item_apply_ms, child_total_ms=child_total_ms, total_ms=_elapsed_ms(started_at), ) return _attach_builder_apply_envelope(tool_name, payload) if not isinstance(apps, list) or not apps: return finish( _config_failure( tool_name=tool_name, message=f"{tool_name} batch mode requires non-empty apps[].", fix_hint="Pass apps as a JSON array; each item must include app_key plus that resource's normal payload.", details={"expected_shape": {"apps": [{"app_key": "APP_KEY"}]}}, ) ) app_results: list[JSONObject] = [] errors: list[JSONObject] = [] for index, item in enumerate(apps): if not isinstance(item, dict): error = { "index": index, "status": "failed", "error_code": "APPS_FILE_ITEM_INVALID", "message": "apps[] item must be an object", "reason_path": f"apps[{index}]", } errors.append(error) app_results.append(error) continue app_key = str(item.get("app_key") or item.get("appKey") or "").strip() if not app_key: error = { "index": index, "status": "failed", "error_code": "APPS_FILE_APP_KEY_REQUIRED", "message": "apps[] item requires app_key", "reason_path": f"apps[{index}].app_key", } errors.append(error) app_results.append(error) continue try: item_started_at = time.perf_counter() result = apply_one(index, item, app_key) except (QingflowApiError, RuntimeError) as error: batch_item_apply_ms += _elapsed_ms(item_started_at) api_error = _coerce_api_error(error) result = { "status": "failed", "error_code": f"{tool_name.upper()}_BATCH_ITEM_FAILED", "recoverable": True, "message": _public_error_message(f"{tool_name.upper()}_BATCH_ITEM_FAILED", api_error), "app_key": app_key, "normalized_args": {"app_key": app_key}, "details": {"transport_error": _transport_error_payload(api_error)}, "request_id": api_error.request_id, "backend_code": api_error.backend_code, "http_status": None if api_error.http_status == 404 else api_error.http_status, } else: batch_item_apply_ms += _elapsed_ms(item_started_at) if isinstance(result, dict): child_breakdown = result.get("duration_breakdown_ms") child_total = child_breakdown.get("total_ms") if isinstance(child_breakdown, dict) else None if isinstance(child_total, (int, float)) and child_total >= 0: child_total_ms += int(child_total) status = str(result.get("status") or "success") if isinstance(result, dict) else "failed" wrapped: JSONObject = { "index": index, "app_key": app_key, "status": status, "result": result if isinstance(result, dict) else {"status": "failed", "message": str(result)}, } if status == "failed": error = { "index": index, "app_key": app_key, "status": status, "error_code": wrapped["result"].get("error_code"), "message": wrapped["result"].get("message"), } errors.append(error) wrapped["error"] = error app_results.append(wrapped) partial = sum(1 for item in app_results if str(item.get("status") or "") == "partial_success") succeeded = sum(1 for item in app_results if str(item.get("status") or "") in {"success", "partial_success"}) failed = len(app_results) - succeeded status = "failed" if succeeded == 0 else "partial_success" if failed > 0 or partial > 0 else "success" write_executed = any( isinstance(item.get("result"), dict) and bool(item["result"].get("write_executed")) for item in app_results ) payload: JSONObject = { "status": status, "error_code": None if status == "success" else f"{tool_name.upper()}_BATCH_PARTIAL" if succeeded else f"{tool_name.upper()}_BATCH_FAILED", "recoverable": failed > 0 or partial > 0, "message": ( f"applied {tool_name} to {succeeded}/{len(app_results)} apps" if status != "success" else f"applied {tool_name} to {succeeded} apps" ), "normalized_args": {"apps": apps}, "apps": app_results, "errors": errors, "verification": { "batch_verified": failed == 0 and partial == 0, "succeeded": succeeded, "partial": partial, "failed": failed, }, "write_executed": write_executed, "write_succeeded": write_executed and failed == 0 and partial == 0, "safe_to_retry": not write_executed, } return finish(payload) def _read_app_batch(self, *, tool_name: str, profile: str, app_keys: list[str], read_one) -> JSONObject: normalized_keys = [str(item).strip() for item in app_keys if str(item).strip()] if not normalized_keys: return _config_failure( tool_name=tool_name, message=f"{tool_name} batch mode requires non-empty app_keys[].", fix_hint="Pass app_keys as a non-empty list of app keys.", details={"expected_shape": {"app_keys": ["APP_A", "APP_B"]}}, ) apps: list[JSONObject] = [] errors: list[JSONObject] = [] for index, app_key in enumerate(normalized_keys): try: result = read_one(app_key) except (QingflowApiError, RuntimeError) as error: api_error = _coerce_api_error(error) result = { "status": "failed", "error_code": f"{tool_name.upper()}_BATCH_ITEM_FAILED", "recoverable": True, "message": _public_error_message(f"{tool_name.upper()}_BATCH_ITEM_FAILED", api_error), "app_key": app_key, "details": {"transport_error": _transport_error_payload(api_error)}, "request_id": api_error.request_id, "backend_code": api_error.backend_code, "http_status": None if api_error.http_status == 404 else api_error.http_status, } status = str(result.get("status") or "success") if isinstance(result, dict) else "failed" app_item: JSONObject = { "index": index, "app_key": app_key, "status": status, "data": result, } if status == "failed": error = { "index": index, "app_key": app_key, "status": "failed", "error_code": result.get("error_code") if isinstance(result, dict) else None, "message": result.get("message") if isinstance(result, dict) else str(result), } app_item["error"] = error errors.append(error) apps.append(app_item) partial = sum(1 for item in apps if str(item.get("status") or "") == "partial_success") succeeded = sum(1 for item in apps if str(item.get("status") or "") in {"success", "partial_success"}) failed = len(apps) - succeeded status = "failed" if succeeded == 0 else "partial_success" if failed > 0 or partial > 0 else "success" return { "status": status, "error_code": None if status == "success" else f"{tool_name.upper()}_BATCH_PARTIAL" if succeeded else f"{tool_name.upper()}_BATCH_FAILED", "recoverable": failed > 0 or partial > 0, "message": f"read {tool_name} for {succeeded}/{len(apps)} apps" if status != "success" else f"read {tool_name} for {succeeded} apps", "normalized_args": {"app_keys": normalized_keys}, "app_keys": normalized_keys, "apps": apps, "errors": errors, "verification": {"batch_verified": failed == 0 and partial == 0, "succeeded": succeeded, "partial": partial, "failed": failed}, } def register(self, mcp) -> None: """注册当前工具到 MCP 服务。""" @mcp.tool() def builder_tool_contract(tool_name: str = "") -> JSONObject: return self.builder_tool_contract(tool_name=tool_name) @mcp.tool() def workspace_icon_catalog_get(profile: str = DEFAULT_PROFILE) -> JSONObject: return self.workspace_icon_catalog_get(profile=profile) @mcp.tool() def package_list(profile: str = DEFAULT_PROFILE, trial_status: str = "all", query: str = "") -> JSONObject: return self.package_list(profile=profile, trial_status=trial_status, query=query) @mcp.tool() def package_get(profile: str = DEFAULT_PROFILE, package_id: int = 0) -> JSONObject: return self.package_get(profile=profile, package_id=package_id) @mcp.tool() def package_apply( profile: str = DEFAULT_PROFILE, package_id: int | None = None, package_name: str | None = None, icon: str | JSONObject | None = None, color: str | None = None, icon_name: str | None = None, icon_color: str | None = None, icon_config: JSONObject | None = None, visibility: JSONObject | None = None, items: list[dict] | None = None, allow_detach: bool = False, ) -> JSONObject: return self.package_apply( profile=profile, package_id=package_id, package_name=package_name, icon=icon, color=color, icon_name=icon_name, icon_color=icon_color, icon_config=icon_config, visibility=visibility, items=items or None, allow_detach=allow_detach, ) @mcp.tool() def solution_install( profile: str = DEFAULT_PROFILE, solution_key: str = "", being_copy_data: bool = True, solution_source: str = "solutionDetail", ) -> JSONObject: return self.solution_install( profile=profile, solution_key=solution_key, being_copy_data=being_copy_data, solution_source=solution_source, ) @mcp.tool() def member_search( profile: str = DEFAULT_PROFILE, query: str = "", page_num: int = 1, page_size: int = 20, contain_disable: bool = False, ) -> JSONObject: return self.member_search( profile=profile, query=query, page_num=page_num, page_size=page_size, contain_disable=contain_disable, ) @mcp.tool() def role_search( profile: str = DEFAULT_PROFILE, keyword: str = "", page_num: int = 1, page_size: int = 20, ) -> JSONObject: return self.role_search(profile=profile, keyword=keyword, page_num=page_num, page_size=page_size) @mcp.tool() def role_create( profile: str = DEFAULT_PROFILE, role_name: str = "", member_uids: list[int] | None = None, member_emails: list[str] | None = None, member_names: list[str] | None = None, role_icon: str = "ex-user-outlined", ) -> JSONObject: return self.role_create( profile=profile, role_name=role_name, member_uids=member_uids or [], member_emails=member_emails or [], member_names=member_names or [], role_icon=role_icon, ) @mcp.tool() def app_release_edit_lock_if_mine( profile: str = DEFAULT_PROFILE, app_key: str = "", lock_owner_email: str = "", lock_owner_name: str = "", ) -> JSONObject: return self.app_release_edit_lock_if_mine( profile=profile, app_key=app_key, lock_owner_email=lock_owner_email, lock_owner_name=lock_owner_name, ) @mcp.tool() def app_resolve( profile: str = DEFAULT_PROFILE, app_key: str = "", app_name: str = "", package_id: int | None = None, ) -> JSONObject: has_app_key = bool((app_key or "").strip()) has_app_name = bool((app_name or "").strip()) has_package_id = package_id is not None if has_app_key and (has_app_name or has_package_id): return _config_failure( tool_name="app_resolve", message="app_resolve accepts exactly one selector mode.", fix_hint="Use only `app_key`, or use `app_name` together with `package_id`.", ) if not has_app_key and not (has_app_name and has_package_id): return _config_failure( tool_name="app_resolve", message="app_resolve requires either app_key, or app_name together with package_id.", fix_hint="For an existing known app, pass `app_key`. For package-scoped lookup, pass both `app_name` and `package_id`.", ) return self.app_resolve(profile=profile, app_key=app_key, app_name=app_name, package_id=package_id) @mcp.tool() def button_style_catalog_get(profile: str = DEFAULT_PROFILE) -> JSONObject: return self.button_style_catalog_get(profile=profile) @mcp.tool() def app_custom_buttons_apply( profile: str = DEFAULT_PROFILE, app_key: str = "", upsert_buttons: list[JSONObject] | None = None, patch_buttons: list[JSONObject] | None = None, remove_buttons: list[JSONObject] | None = None, view_configs: list[JSONObject] | None = None, apps: list[JSONObject] | None = None, ) -> JSONObject: return self.app_custom_buttons_apply( profile=profile, app_key=app_key, upsert_buttons=upsert_buttons or [], patch_buttons=patch_buttons or [], remove_buttons=remove_buttons or [], view_configs=view_configs or [], apps=apps, ) @mcp.tool() def app_associated_resources_apply( profile: str = DEFAULT_PROFILE, app_key: str = "", upsert_resources: list[JSONObject] | None = None, patch_resources: list[JSONObject] | None = None, remove_associated_item_ids: list[int] | None = None, reorder_associated_item_ids: list[int] | None = None, view_configs: list[JSONObject] | None = None, apps: list[JSONObject] | None = None, ) -> JSONObject: return self.app_associated_resources_apply( profile=profile, app_key=app_key, upsert_resources=upsert_resources or [], patch_resources=patch_resources or [], remove_associated_item_ids=remove_associated_item_ids or [], reorder_associated_item_ids=reorder_associated_item_ids or [], view_configs=view_configs or [], apps=apps, ) @mcp.tool() def app_get(profile: str = DEFAULT_PROFILE, app_key: str = "", app_keys: list[str] | None = None) -> JSONObject: return self.app_get(profile=profile, app_key=app_key, app_keys=app_keys) @mcp.tool() def app_get_fields(profile: str = DEFAULT_PROFILE, app_key: str = "", app_keys: list[str] | None = None) -> JSONObject: return self.app_get_fields(profile=profile, app_key=app_key, app_keys=app_keys) @mcp.tool() def app_repair_code_blocks( profile: str = DEFAULT_PROFILE, app_key: str = "", field: str | None = None, apply: bool = False, ) -> JSONObject: return self.app_repair_code_blocks(profile=profile, app_key=app_key, field=field, apply=apply) @mcp.tool() def app_get_layout(profile: str = DEFAULT_PROFILE, app_key: str = "", app_keys: list[str] | None = None) -> JSONObject: return self.app_get_layout(profile=profile, app_key=app_key, app_keys=app_keys) @mcp.tool() def app_get_views(profile: str = DEFAULT_PROFILE, app_key: str = "", app_keys: list[str] | None = None) -> JSONObject: return self.app_get_views(profile=profile, app_key=app_key, app_keys=app_keys) @mcp.tool() def app_get_flow(profile: str = DEFAULT_PROFILE, app_key: str = "", app_keys: list[str] | None = None) -> JSONObject: return self.app_get_flow(profile=profile, app_key=app_key, app_keys=app_keys) @mcp.tool() def app_get_charts(profile: str = DEFAULT_PROFILE, app_key: str = "", app_keys: list[str] | None = None) -> JSONObject: return self.app_get_charts(profile=profile, app_key=app_key, app_keys=app_keys) @mcp.tool() def app_get_buttons(profile: str = DEFAULT_PROFILE, app_key: str = "", app_keys: list[str] | None = None) -> JSONObject: return self.app_get_buttons(profile=profile, app_key=app_key, app_keys=app_keys) @mcp.tool() def app_get_associated_resources(profile: str = DEFAULT_PROFILE, app_key: str = "", app_keys: list[str] | None = None) -> JSONObject: return self.app_get_associated_resources(profile=profile, app_key=app_key, app_keys=app_keys) @mcp.tool() def portal_list(profile: str = DEFAULT_PROFILE) -> JSONObject: return self.portal_list(profile=profile) @mcp.tool() def portal_get( profile: str = DEFAULT_PROFILE, dash_key: str = "", being_draft: bool = True, ) -> JSONObject: return self.portal_get(profile=profile, dash_key=dash_key, being_draft=being_draft) @mcp.tool() def view_get(profile: str = DEFAULT_PROFILE, view_key: str = "") -> JSONObject: return self.view_get(profile=profile, view_key=view_key) @mcp.tool() def chart_get( profile: str = DEFAULT_PROFILE, chart_id: str = "", ) -> JSONObject: return self.chart_get(profile=profile, chart_id=chart_id) @mcp.tool() def app_schema_apply( profile: str = DEFAULT_PROFILE, app_key: str = "", package_id: int | None = None, app_name: str = "", app_title: str = "", icon: str | JSONObject = "", color: str = "", icon_name: str | None = None, icon_color: str | None = None, icon_config: JSONObject | None = None, visibility: JSONObject | None = None, publish: bool = True, form: list[JSONObject] | None = None, add_fields: list[JSONObject] | None = None, update_fields: list[JSONObject] | None = None, remove_fields: list[JSONObject] | None = None, apps: list[JSONObject] | None = None, ) -> JSONObject: if apps is not None: if app_key or app_name or app_title or form or add_fields or update_fields or remove_fields: return _config_failure( tool_name="app_schema_apply", message="app_schema_apply multi-app mode accepts package_id plus apps only.", fix_hint="Put per-app form in apps[].form when using batch mode.", ) if package_id is None: return _config_failure( tool_name="app_schema_apply", message="app_schema_apply multi-app mode requires package_id.", fix_hint="Pass package_id and apps[].app_name for new apps, or apps[].app_key for existing apps.", ) return self.app_schema_apply( profile=profile, package_id=package_id, visibility=visibility, publish=publish, apps=apps, form=None, add_fields=[], update_fields=[], remove_fields=[], ) has_app_key = bool((app_key or "").strip()) has_app_name = bool((app_name or "").strip()) has_app_title = bool((app_title or "").strip()) has_package_id = package_id is not None if has_app_key: if has_package_id: return _config_failure( tool_name="app_schema_apply", message="app_schema_apply edit mode accepts app_key and optional app_name rename only.", fix_hint="For existing apps, use `app_key` and optionally `app_name`. For create mode, use `package_id + app_name`.", ) elif not (has_package_id and (has_app_name or has_app_title)): return _config_failure( tool_name="app_schema_apply", message="app_schema_apply create mode requires package_id and app_name.", fix_hint="Use `app_key` for existing apps, or pass `package_id + app_name` to create a new app.", ) return self.app_schema_apply( profile=profile, app_key=app_key, package_id=package_id, app_name=app_name, app_title=app_title, icon=icon, color=color, icon_name=icon_name, icon_color=icon_color, icon_config=icon_config, visibility=visibility, publish=publish, form=form or [], add_fields=add_fields or [], update_fields=update_fields or [], remove_fields=remove_fields or [], apps=None, ) @mcp.tool() def app_layout_apply( profile: str = DEFAULT_PROFILE, app_key: str = "", mode: str = "merge", publish: bool = True, sections: list[JSONObject] | None = None, apps: list[JSONObject] | None = None, ) -> JSONObject: return self.app_layout_apply(profile=profile, app_key=app_key, mode=mode, publish=publish, sections=sections or [], apps=apps) @mcp.tool() def app_flow_get( profile: str = DEFAULT_PROFILE, app_key: str = "", version_id: str = "", ) -> JSONObject: return self.app_flow_get(profile=profile, app_key=app_key, version_id=version_id or None) @mcp.tool() def app_flow_get_schema(profile: str = DEFAULT_PROFILE, schema_version: str = "") -> JSONObject: return self.app_flow_get_schema(profile=profile, schema_version=schema_version or None) @mcp.tool() def app_flow_apply( profile: str = DEFAULT_PROFILE, app_key: str = "", publish: bool = True, spec: JSONObject | None = None, idempotency_key: str = "", schema_version: str = "", patch_nodes: list[JSONObject] | None = None, ) -> JSONObject: return self.app_flow_apply( profile=profile, app_key=app_key, publish=publish, spec=spec, idempotency_key=idempotency_key or None, schema_version=schema_version or None, patch_nodes=patch_nodes, ) @mcp.tool() def app_views_apply( profile: str = DEFAULT_PROFILE, app_key: str = "", publish: bool = True, upsert_views: list[JSONObject] | None = None, patch_views: list[JSONObject] | None = None, remove_views: list[str] | None = None, apps: list[JSONObject] | None = None, ) -> JSONObject: return self.app_views_apply( profile=profile, app_key=app_key, publish=publish, upsert_views=upsert_views or [], patch_views=patch_views or [], remove_views=remove_views or [], apps=apps, ) @mcp.tool() def app_charts_apply( profile: str = DEFAULT_PROFILE, app_key: str = "", upsert_charts: list[JSONObject] | None = None, patch_charts: list[JSONObject] | None = None, remove_chart_ids: list[str] | None = None, reorder_chart_ids: list[str] | None = None, apps: list[JSONObject] | None = None, ) -> JSONObject: return self.app_charts_apply( profile=profile, app_key=app_key, upsert_charts=upsert_charts or [], patch_charts=patch_charts or [], remove_chart_ids=remove_chart_ids or [], reorder_chart_ids=reorder_chart_ids or [], apps=apps, ) @mcp.tool() def portal_apply( profile: str = DEFAULT_PROFILE, dash_key: str = "", dash_name: str = "", name: str = "", package_id: int | None = None, publish: bool = True, sections: list[JSONObject] | None = None, pages: list[JSONObject] | None = None, layout_preset: str = "", visibility: JSONObject | None = None, auth: JSONObject | None = None, icon: str | None = None, color: str | None = None, hide_copyright: bool | None = None, dash_global_config: JSONObject | None = None, config: JSONObject | None = None, payload: JSONObject | None = None, patch_sections: list[JSONObject] | None = None, ) -> JSONObject: payload = payload if isinstance(payload, dict) else {} has_dash_key = bool((dash_key or "").strip()) effective_dash_name = (dash_name or name or str(payload.get("dash_name") or payload.get("dashName") or payload.get("name") or "")).strip() has_dash_name = bool(effective_dash_name) effective_package_id = package_id if package_id is not None else payload.get("package_id") or payload.get("packageId") or payload.get("package_tag_id") has_package_id = effective_package_id is not None if has_dash_key and has_package_id: return _config_failure( tool_name="portal_apply", message="portal_apply accepts exactly one selector mode.", fix_hint="Use `dash_key` to update an existing portal, or use `package_id + dash_name` to create a new portal.", ) if not has_dash_key and not (has_package_id and has_dash_name): return _config_failure( tool_name="portal_apply", message="portal_apply requires either dash_key, or package_id together with dash_name.", fix_hint="Use `dash_key` for an existing portal. For create mode, pass `package_id + dash_name`.", ) return self.portal_apply( profile=profile, dash_key=dash_key, dash_name=dash_name, name=name, package_id=package_id, publish=publish, sections=sections or [], pages=pages or [], layout_preset=layout_preset, visibility=visibility, auth=auth, icon=icon, color=color, hide_copyright=hide_copyright, dash_global_config=dash_global_config, config=config or {}, payload=payload, patch_sections=patch_sections, ) @mcp.tool(description=self._high_risk_tool_description(operation="delete", target="portal")) def portal_delete( profile: str = DEFAULT_PROFILE, dash_key: str = "", ) -> JSONObject: return self.portal_delete(profile=profile, dash_key=dash_key) @mcp.tool() def app_publish_verify( profile: str = DEFAULT_PROFILE, app_key: str = "", app_keys: list[str] | None = None, expected_package_id: int | None = None, ) -> JSONObject: return self.app_publish_verify( profile=profile, app_key=app_key, app_keys=app_keys, expected_package_id=expected_package_id, ) def package_list(self, *, profile: str, trial_status: str = "all", query: str = "") -> JSONObject: """执行分组与包相关逻辑。""" normalized_args = {"trial_status": trial_status, "query": query} return _safe_tool_call( lambda: self._facade.package_list(profile=profile, trial_status=trial_status, query=query), error_code="PACKAGE_LIST_FAILED", normalized_args=normalized_args, suggested_next_call={"tool_name": "package_list", "arguments": {"profile": profile, "trial_status": trial_status, "query": query}}, ) def package_resolve(self, *, profile: str, package_name: str) -> JSONObject: """执行分组与包相关逻辑。""" normalized_args = {"package_name": package_name} return _safe_tool_call( lambda: self._facade.package_resolve(profile=profile, package_name=package_name), error_code="PACKAGE_RESOLVE_FAILED", normalized_args=normalized_args, suggested_next_call={"tool_name": "package_resolve", "arguments": {"profile": profile, "package_name": package_name}}, ) def builder_tool_contract(self, *, tool_name: str) -> JSONObject: """执行工具方法逻辑。""" requested = str(tool_name or "").strip() public_tool_names = public_builder_contract_tool_names() if requested in _PRIVATE_BUILDER_TOOL_CONTRACTS: lookup_name = "" else: lookup_name = _BUILDER_TOOL_CONTRACT_ALIASES.get(requested, requested) if lookup_name not in public_tool_names: lookup_name = "" contract = _BUILDER_TOOL_CONTRACTS.get(lookup_name) if contract is None: return { "status": "failed", "error_code": "TOOL_CONTRACT_NOT_FOUND", "recoverable": True, "message": "tool contract is not defined for the requested public builder tool", "normalized_args": {"tool_name": requested}, "missing_fields": [], "allowed_values": {"tool_name": public_tool_names}, "details": {"reason_path": "tool_name"}, "suggested_next_call": None, "request_id": None, "backend_code": None, "http_status": None, "noop": False, "warnings": [], "verification": {}, "verified": False, } contract = _builder_contract_with_apply_output(lookup_name, contract) contract_summary = _builder_tool_contract_summary(lookup_name, contract) return { "status": "success", "error_code": None, "recoverable": False, "message": "loaded builder tool contract", "normalized_args": {"tool_name": requested}, "missing_fields": [], "allowed_values": {}, "details": {}, "suggested_next_call": None, "request_id": None, "backend_code": None, "http_status": None, "noop": False, "warnings": [], "verification": {}, "verified": True, "tool_name": requested, "summary": contract_summary, "json_paths": contract_summary["json_paths"], "contract": contract, } def workspace_icon_catalog_get(self, *, profile: str = DEFAULT_PROFILE) -> JSONObject: """读取应用、应用包、门户可用的工作区图标候选。""" catalog = workspace_icon_catalog_payload() return { "status": "success", "error_code": None, "recoverable": False, "message": "loaded workspace icon catalog", "profile": profile, "icon_names": catalog["icon_names"], "icon_colors": catalog["icon_colors"], "generic_icon_names": catalog["generic_icon_names"], "common_examples": catalog["common_examples"], "notes": catalog["notes"], "count": len(catalog["icon_names"]), "color_count": len(catalog["icon_colors"]), "warnings": [], "verification": {"source": "backend AiBuildConstant ICON_NAMES/ICON_COLORS"}, "json_paths": { "icon_names": "$.icon_names", "icon_colors": "$.icon_colors", "generic_icon_names": "$.generic_icon_names", "common_examples": "$.common_examples", }, } def package_create( self, *, profile: str, package_name: str, icon: str | None = None, color: str | None = None, visibility: JSONObject | None = None, ) -> JSONObject: """执行分组与包相关逻辑。""" visibility_patch = None if visibility is not None: try: visibility_patch = VisibilityPatch.model_validate(visibility) except ValidationError as exc: return _visibility_validation_failure(str(exc), tool_name="package_create", exc=exc) normalized_args = { "package_name": package_name, **({"icon": icon} if icon else {}), **({"color": color} if color else {}), **({"visibility": visibility_patch.model_dump(mode="json")} if visibility_patch is not None else {}), } return _safe_tool_call( lambda: self._facade.package_create( profile=profile, package_name=package_name, icon=icon, color=color, visibility=visibility_patch, ), error_code="PACKAGE_CREATE_FAILED", normalized_args=normalized_args, suggested_next_call={ "tool_name": "package_create", "arguments": { "profile": profile, "package_name": package_name, **({"icon": icon} if icon else {}), **({"color": color} if color else {}), **({"visibility": visibility_patch.model_dump(mode="json")} if visibility_patch is not None else {}), }, }, ) def package_get(self, *, profile: str, package_id: int | None = None) -> JSONObject: """执行分组与包相关逻辑。""" normalized_args = {"package_id": package_id} return _publicize_package_fields(_safe_tool_call( lambda: self._facade.package_get(profile=profile, package_id=package_id), error_code="PACKAGE_GET_FAILED", normalized_args=normalized_args, suggested_next_call={"tool_name": "package_get", "arguments": {"profile": profile, "package_id": package_id}}, )) def package_apply( self, *, profile: str, package_id: int | None = None, package_name: str | None = None, icon: str | JSONObject | None = None, color: str | None = None, icon_name: str | None = None, icon_color: str | None = None, icon_config: JSONObject | None = None, visibility: JSONObject | None = None, items: list[dict] | None = None, allow_detach: bool = False, ) -> JSONObject: """执行分组与包相关逻辑。""" icon, color = _normalize_builder_icon_args( icon=icon, color=color, icon_name=icon_name, icon_color=icon_color, icon_config=icon_config, ) visibility_patch = None if visibility is not None: try: visibility_patch = VisibilityPatch.model_validate(visibility) except ValidationError as exc: return _attach_builder_apply_envelope( "package_apply", _visibility_validation_failure(str(exc), tool_name="package_apply", exc=exc), ) icon_failure = _validate_workspace_icon_for_builder( tool_name="package_apply", icon=icon, color=color, creating=package_id is None and bool(str(package_name or "").strip()), ) if icon_failure is not None: return _attach_builder_apply_envelope("package_apply", icon_failure) normalized_args = { "package_id": package_id, **({"package_name": package_name} if str(package_name or "").strip() else {}), **({"icon": icon} if icon else {}), **({"color": color} if color else {}), **({"visibility": visibility_patch.model_dump(mode="json")} if visibility_patch is not None else {}), **({"items": deepcopy(items)} if items is not None else {}), "allow_detach": bool(allow_detach), } result = _publicize_package_fields(_safe_tool_call( lambda: self._facade.package_apply( profile=profile, package_id=package_id, package_name=package_name, icon=icon, color=color, visibility=visibility_patch, items=items, allow_detach=allow_detach, ), error_code="PACKAGE_APPLY_FAILED", normalized_args=normalized_args, suggested_next_call={"tool_name": "package_apply", "arguments": {"profile": profile, **normalized_args}}, )) return _attach_builder_apply_envelope("package_apply", result) def package_update( self, *, profile: str, tag_id: int, package_name: str | None = None, icon: str | None = None, color: str | None = None, visibility: JSONObject | None = None, ) -> JSONObject: """执行分组与包相关逻辑。""" visibility_patch = None if visibility is not None: try: visibility_patch = VisibilityPatch.model_validate(visibility) except ValidationError as exc: return _visibility_validation_failure(str(exc), tool_name="package_update", exc=exc) normalized_args = { "tag_id": tag_id, **({"package_name": package_name} if str(package_name or "").strip() else {}), **({"icon": icon} if icon else {}), **({"color": color} if color else {}), **({"visibility": visibility_patch.model_dump(mode="json")} if visibility_patch is not None else {}), } return _safe_tool_call( lambda: self._facade.package_update( profile=profile, tag_id=tag_id, package_name=package_name, icon=icon, color=color, visibility=visibility_patch, ), error_code="PACKAGE_UPDATE_FAILED", normalized_args=normalized_args, suggested_next_call={"tool_name": "package_update", "arguments": {"profile": profile, **normalized_args}}, ) def solution_install( self, *, profile: str, solution_key: str, being_copy_data: bool = True, solution_source: str = "solutionDetail", ) -> JSONObject: """执行方案相关逻辑。""" normalized_args = { "solution_key": solution_key, "being_copy_data": being_copy_data, "solution_source": solution_source, } return _safe_tool_call( lambda: self._facade.solution_install( profile=profile, solution_key=solution_key, being_copy_data=being_copy_data, solution_source=solution_source, ), error_code="SOLUTION_INSTALL_FAILED", normalized_args=normalized_args, suggested_next_call={"tool_name": "solution_install", "arguments": {"profile": profile, **normalized_args}}, ) def member_search( self, *, profile: str, query: str, page_num: int = 1, page_size: int = 20, contain_disable: bool = False, ) -> JSONObject: """执行工具方法逻辑。""" normalized_query = str(query or "").strip() normalized_args = { "query": normalized_query, "page_num": page_num, "page_size": page_size, "contain_disable": contain_disable, } if not normalized_query: return _config_failure( tool_name="member_search", message="query is required for member_search; builder member lookup is a contact-directory path, not a record candidate fallback.", fix_hint="For record member/department field ambiguity, use record member-candidates / department-candidates instead.", ) return _safe_tool_call( lambda: self._facade.member_search( profile=profile, query=normalized_query, page_num=page_num, page_size=page_size, contain_disable=contain_disable, ), error_code="MEMBER_SEARCH_FAILED", normalized_args=normalized_args, suggested_next_call={"tool_name": "member_search", "arguments": {"profile": profile, **normalized_args}}, ) def role_search(self, *, profile: str, keyword: str, page_num: int = 1, page_size: int = 20) -> JSONObject: """执行角色相关逻辑。""" normalized_keyword = str(keyword or "").strip() normalized_args = {"keyword": normalized_keyword, "page_num": page_num, "page_size": page_size} if not normalized_keyword: return _config_failure( tool_name="role_search", message="keyword is required for role_search; builder role lookup is a contact-management path, not a record candidate fallback.", fix_hint="For record member/department field ambiguity, use record member-candidates / department-candidates instead.", ) return _safe_tool_call( lambda: self._facade.role_search(profile=profile, keyword=normalized_keyword, page_num=page_num, page_size=page_size), error_code="ROLE_SEARCH_FAILED", normalized_args=normalized_args, suggested_next_call={"tool_name": "role_search", "arguments": {"profile": profile, **normalized_args}}, ) def role_create( self, *, profile: str, role_name: str, member_uids: list[int], member_emails: list[str], member_names: list[str], role_icon: str = "ex-user-outlined", ) -> JSONObject: """执行角色相关逻辑。""" normalized_args = { "role_name": role_name, "member_uids": member_uids, "member_emails": member_emails, "member_names": member_names, "role_icon": role_icon, } return _safe_tool_call( lambda: self._facade.role_create( profile=profile, role_name=role_name, member_uids=member_uids, member_emails=member_emails, member_names=member_names, role_icon=role_icon, ), error_code="ROLE_CREATE_FAILED", normalized_args=normalized_args, suggested_next_call={"tool_name": "role_create", "arguments": {"profile": profile, **normalized_args}}, ) def package_attach_app(self, *, profile: str, tag_id: int, app_key: str, app_title: str = "") -> JSONObject: """执行分组与包相关逻辑。""" normalized_args = {"tag_id": tag_id, "app_key": app_key, "app_title": app_title} result = _safe_tool_call( lambda: self._facade.package_attach_app(profile=profile, tag_id=tag_id, app_key=app_key, app_title=app_title), error_code="PACKAGE_ATTACH_FAILED", normalized_args=normalized_args, suggested_next_call={"tool_name": "package_attach_app", "arguments": {"profile": profile, **normalized_args}}, ) return self._retry_after_self_lock_release( profile=profile, result=result, retry_call=lambda: self._facade.package_attach_app( profile=profile, tag_id=tag_id, app_key=app_key, app_title=app_title, ), ) def app_release_edit_lock_if_mine( self, *, profile: str, app_key: str, lock_owner_email: str = "", lock_owner_name: str = "", ) -> JSONObject: """执行应用相关逻辑。""" normalized_args = { "app_key": app_key, "lock_owner_email": lock_owner_email, "lock_owner_name": lock_owner_name, } return _safe_tool_call( lambda: self._facade.app_release_edit_lock_if_mine( profile=profile, app_key=app_key, lock_owner_email=lock_owner_email, lock_owner_name=lock_owner_name, ), error_code="EDIT_LOCK_RELEASE_FAILED", normalized_args=normalized_args, suggested_next_call={"tool_name": "app_release_edit_lock_if_mine", "arguments": {"profile": profile, **normalized_args}}, ) def app_resolve( self, *, profile: str, app_key: str = "", app_name: str = "", package_id: int | None = None, ) -> JSONObject: """执行应用相关逻辑。""" normalized_args = {"app_key": app_key, "app_name": app_name, "package_id": package_id} return _publicize_package_fields(_safe_tool_call( lambda: self._facade.app_resolve(profile=profile, app_key=app_key, app_name=app_name, package_tag_id=package_id), error_code="APP_RESOLVE_FAILED", normalized_args=normalized_args, suggested_next_call={"tool_name": "app_resolve", "arguments": {"profile": profile, **normalized_args}}, )) def button_style_catalog_get(self, *, profile: str) -> JSONObject: """执行按钮样式相关逻辑。""" normalized_args: dict[str, object] = {} return _safe_tool_call( lambda: self._facade.button_style_catalog_get(profile=profile), error_code="BUTTON_STYLE_CATALOG_GET_FAILED", normalized_args=normalized_args, suggested_next_call={"tool_name": "button_style_catalog_get", "arguments": {"profile": profile}}, ) def app_custom_buttons_apply( self, *, profile: str, app_key: str, upsert_buttons: list[JSONObject], patch_buttons: list[JSONObject] | None = None, remove_buttons: list[JSONObject], view_configs: list[JSONObject] | None = None, apps: list[JSONObject] | None = None, ) -> JSONObject: """执行应用按钮 apply 逻辑。""" if apps is not None: return self._apply_app_batch( tool_name="app_custom_buttons_apply", profile=profile, apps=apps, apply_one=lambda _index, item, item_app_key: self.app_custom_buttons_apply( profile=profile, app_key=item_app_key, upsert_buttons=_payload_list(item, "upsert_buttons", "upsertButtons", "buttons", default=upsert_buttons or []), patch_buttons=_payload_list(item, "patch_buttons", "patchButtons", default=patch_buttons or []), remove_buttons=_payload_list(item, "remove_buttons", "removeButtons", default=remove_buttons or []), view_configs=_payload_list(item, "view_configs", "viewConfigs", default=view_configs or []), apps=None, ), ) raw_request = { "app_key": app_key, "upsert_buttons": upsert_buttons, "patch_buttons": patch_buttons or [], "remove_buttons": remove_buttons, "view_configs": view_configs or [], } try: request = CustomButtonsApplyRequest.model_validate(raw_request) except ValidationError as exc: return _attach_builder_apply_envelope("app_custom_buttons_apply", _validation_failure( str(exc), tool_name="app_custom_buttons_apply", exc=exc, suggested_next_call={ "tool_name": "app_custom_buttons_apply", "arguments": { "profile": profile, "app_key": app_key or "APP_KEY", "upsert_buttons": [ { "button_text": "同步客户", "style_preset": "primary_blue", "button_icon": "ex-switch", "trigger_action": "link", "trigger_link_url": "https://example.com", } ], "remove_buttons": [], "view_configs": [], }, }, )) normalized_args = request.model_dump(mode="json") return _attach_builder_apply_envelope("app_custom_buttons_apply", _safe_tool_call( lambda: self._facade.app_custom_buttons_apply(profile=profile, request=request), error_code="CUSTOM_BUTTONS_APPLY_FAILED", normalized_args=normalized_args, suggested_next_call={"tool_name": "app_custom_buttons_apply", "arguments": {"profile": profile, **normalized_args}}, )) def app_associated_resources_apply( self, *, profile: str, app_key: str, upsert_resources: list[JSONObject], remove_associated_item_ids: list[int], reorder_associated_item_ids: list[int], view_configs: list[JSONObject], patch_resources: list[JSONObject] | None = None, apps: list[JSONObject] | None = None, ) -> JSONObject: """执行应用关联资源 apply 逻辑。""" if apps is not None: return self._apply_app_batch( tool_name="app_associated_resources_apply", profile=profile, apps=apps, apply_one=lambda _index, item, item_app_key: self.app_associated_resources_apply( profile=profile, app_key=item_app_key, upsert_resources=_payload_list(item, "upsert_resources", "upsertResources", "resources", default=upsert_resources or []), patch_resources=_payload_list(item, "patch_resources", "patchResources", default=patch_resources or []), remove_associated_item_ids=_payload_list( item, "remove_associated_item_ids", "removeAssociatedItemIds", default=remove_associated_item_ids or [], ), reorder_associated_item_ids=_payload_list( item, "reorder_associated_item_ids", "reorderAssociatedItemIds", default=reorder_associated_item_ids or [], ), view_configs=_payload_list(item, "view_configs", "viewConfigs", default=view_configs or []), apps=None, ), ) raw_request = { "app_key": app_key, "upsert_resources": upsert_resources, "patch_resources": patch_resources or [], "remove_associated_item_ids": remove_associated_item_ids, "reorder_associated_item_ids": reorder_associated_item_ids, "view_configs": view_configs, } try: request = AssociatedResourcesApplyRequest.model_validate(raw_request) except ValidationError as exc: return _attach_builder_apply_envelope("app_associated_resources_apply", _validation_failure( str(exc), tool_name="app_associated_resources_apply", exc=exc, suggested_next_call={ "tool_name": "app_associated_resources_apply", "arguments": { "profile": profile, "app_key": app_key or "APP_KEY", "upsert_resources": [ { "client_key": "customer_view", "graph_type": "view", "target_app_key": "TARGET_APP", "view_key": "VIEW_KEY", } ], "view_configs": [ { "view_key": "MAIN_VIEW", "visible": True, "limit_type": "select", "associated_item_refs": ["customer_view"], } ], }, }, )) normalized_args = request.model_dump(mode="json", exclude_none=True) return _attach_builder_apply_envelope("app_associated_resources_apply", _safe_tool_call( lambda: self._facade.app_associated_resources_apply(profile=profile, request=request), error_code="ASSOCIATED_RESOURCES_APPLY_FAILED", normalized_args=normalized_args, suggested_next_call={"tool_name": "app_associated_resources_apply", "arguments": {"profile": profile, **normalized_args}}, )) def app_custom_button_list(self, *, profile: str, app_key: str) -> JSONObject: """执行应用相关逻辑。""" normalized_args = {"app_key": app_key} return _safe_tool_call( lambda: self._facade.app_custom_button_list(profile=profile, app_key=app_key), error_code="CUSTOM_BUTTON_LIST_FAILED", normalized_args=normalized_args, suggested_next_call={"tool_name": "app_custom_button_list", "arguments": {"profile": profile, **normalized_args}}, ) def app_custom_button_get(self, *, profile: str, app_key: str, button_id: int) -> JSONObject: """执行应用相关逻辑。""" normalized_args = {"app_key": app_key, "button_id": button_id} return _safe_tool_call( lambda: self._facade.app_custom_button_get(profile=profile, app_key=app_key, button_id=button_id), error_code="CUSTOM_BUTTON_GET_FAILED", normalized_args=normalized_args, suggested_next_call={"tool_name": "app_custom_button_get", "arguments": {"profile": profile, **normalized_args}}, ) def app_custom_button_create(self, *, profile: str, app_key: str, payload: JSONObject) -> JSONObject: """执行应用相关逻辑。""" try: request = CustomButtonPatch.model_validate(payload) except ValidationError as exc: return _validation_failure( str(exc), tool_name="app_custom_button_create", exc=exc, suggested_next_call={ "tool_name": "app_custom_button_create", "arguments": { "profile": profile, "app_key": app_key, "payload": { "button_text": "新增记录", "style_preset": "primary_blue", "button_icon": "ex-plus-circle", "trigger_action": "addData", "trigger_add_data_config": { "target_app_key": "TARGET_APP_KEY", "field_mappings": [{"source_field": "客户名称", "target_field": "客户"}], }, }, }, }, ) normalized_args = {"app_key": app_key, "payload": request.model_dump(mode="json")} return _safe_tool_call( lambda: self._facade.app_custom_button_create(profile=profile, app_key=app_key, payload=request), error_code="CUSTOM_BUTTON_CREATE_FAILED", normalized_args=normalized_args, suggested_next_call={"tool_name": "app_custom_button_create", "arguments": {"profile": profile, **normalized_args}}, ) def app_custom_button_update(self, *, profile: str, app_key: str, button_id: int, payload: JSONObject) -> JSONObject: """执行应用相关逻辑。""" try: request = CustomButtonPatch.model_validate(payload) except ValidationError as exc: return _validation_failure( str(exc), tool_name="app_custom_button_update", exc=exc, suggested_next_call={ "tool_name": "app_custom_button_update", "arguments": { "profile": profile, "app_key": app_key, "button_id": button_id, "payload": { "button_text": "新增记录", "style_preset": "neutral_outline", "button_icon": "ex-edit", "trigger_action": "link", "trigger_link_url": "https://example.com", }, }, }, ) normalized_args = {"app_key": app_key, "button_id": button_id, "payload": request.model_dump(mode="json")} return _safe_tool_call( lambda: self._facade.app_custom_button_update(profile=profile, app_key=app_key, button_id=button_id, payload=request), error_code="CUSTOM_BUTTON_UPDATE_FAILED", normalized_args=normalized_args, suggested_next_call={"tool_name": "app_custom_button_update", "arguments": {"profile": profile, **normalized_args}}, ) def app_custom_button_delete(self, *, profile: str, app_key: str, button_id: int) -> JSONObject: """执行应用相关逻辑。""" normalized_args = {"app_key": app_key, "button_id": button_id} return _safe_tool_call( lambda: self._facade.app_custom_button_delete(profile=profile, app_key=app_key, button_id=button_id), error_code="CUSTOM_BUTTON_DELETE_FAILED", normalized_args=normalized_args, suggested_next_call={"tool_name": "app_custom_button_delete", "arguments": {"profile": profile, **normalized_args}}, ) def app_read_summary(self, *, profile: str, app_key: str) -> JSONObject: """执行应用相关逻辑。""" normalized_args = {"app_key": app_key} return _publicize_package_fields(_safe_tool_call( lambda: self._facade.app_read_summary(profile=profile, app_key=app_key), error_code="APP_READ_FAILED", normalized_args=normalized_args, suggested_next_call={"tool_name": "app_get", "arguments": {"profile": profile, "app_key": app_key}}, )) def app_get(self, *, profile: str, app_key: str, app_keys: list[str] | None = None) -> JSONObject: """执行应用相关逻辑。""" if app_keys is not None: return self._read_app_batch( tool_name="app_get", profile=profile, app_keys=app_keys, read_one=lambda item_app_key: self.app_get(profile=profile, app_key=item_app_key, app_keys=None), ) normalized_args = {"app_key": app_key} return _publicize_package_fields(_safe_tool_call( lambda: self._facade.app_get(profile=profile, app_key=app_key), error_code="APP_GET_FAILED", normalized_args=normalized_args, suggested_next_call={"tool_name": "app_get", "arguments": {"profile": profile, "app_key": app_key}}, )) def app_read_fields(self, *, profile: str, app_key: str) -> JSONObject: """执行应用相关逻辑。""" normalized_args = {"app_key": app_key} return _safe_tool_call( lambda: self._facade.app_read_fields(profile=profile, app_key=app_key), error_code="FIELDS_READ_FAILED", normalized_args=normalized_args, suggested_next_call={"tool_name": "app_get_fields", "arguments": {"profile": profile, "app_key": app_key}}, ) def app_get_fields(self, *, profile: str, app_key: str, app_keys: list[str] | None = None) -> JSONObject: """执行应用相关逻辑。""" if app_keys is not None: return self._read_app_batch( tool_name="app_get_fields", profile=profile, app_keys=app_keys, read_one=lambda item_app_key: self.app_get_fields(profile=profile, app_key=item_app_key, app_keys=None), ) normalized_args = {"app_key": app_key} return _safe_tool_call( lambda: self._facade.app_get_fields(profile=profile, app_key=app_key), error_code="APP_GET_FIELDS_FAILED", normalized_args=normalized_args, suggested_next_call={"tool_name": "app_get_fields", "arguments": {"profile": profile, "app_key": app_key}}, ) def app_repair_code_blocks( self, *, profile: str, app_key: str, field: str | None = None, apply: bool = False, ) -> JSONObject: """执行应用相关逻辑。""" normalized_args = {"app_key": app_key, "field": field, "apply": apply} return _safe_tool_call( lambda: self._facade.app_repair_code_blocks(profile=profile, app_key=app_key, field=field, apply=apply), error_code="APP_REPAIR_CODE_BLOCKS_FAILED", normalized_args=normalized_args, suggested_next_call={"tool_name": "app_repair_code_blocks", "arguments": {"profile": profile, **normalized_args}}, ) def app_read_layout_summary(self, *, profile: str, app_key: str) -> JSONObject: """执行应用相关逻辑。""" normalized_args = {"app_key": app_key} return _safe_tool_call( lambda: self._facade.app_read_layout_summary(profile=profile, app_key=app_key), error_code="LAYOUT_READ_FAILED", normalized_args=normalized_args, suggested_next_call={"tool_name": "app_get_layout", "arguments": {"profile": profile, "app_key": app_key}}, ) def app_get_layout(self, *, profile: str, app_key: str, app_keys: list[str] | None = None) -> JSONObject: """执行应用相关逻辑。""" if app_keys is not None: return self._read_app_batch( tool_name="app_get_layout", profile=profile, app_keys=app_keys, read_one=lambda item_app_key: self.app_get_layout(profile=profile, app_key=item_app_key, app_keys=None), ) normalized_args = {"app_key": app_key} return _safe_tool_call( lambda: self._facade.app_get_layout(profile=profile, app_key=app_key), error_code="APP_GET_LAYOUT_FAILED", normalized_args=normalized_args, suggested_next_call={"tool_name": "app_get_layout", "arguments": {"profile": profile, "app_key": app_key}}, ) def app_read_views_summary(self, *, profile: str, app_key: str) -> JSONObject: """执行应用相关逻辑。""" normalized_args = {"app_key": app_key} return _safe_tool_call( lambda: self._facade.app_read_views_summary(profile=profile, app_key=app_key), error_code="VIEWS_READ_FAILED", normalized_args=normalized_args, suggested_next_call={"tool_name": "app_get_views", "arguments": {"profile": profile, "app_key": app_key}}, ) def app_get_views(self, *, profile: str, app_key: str, app_keys: list[str] | None = None) -> JSONObject: """执行应用相关逻辑。""" if app_keys is not None: return self._read_app_batch( tool_name="app_get_views", profile=profile, app_keys=app_keys, read_one=lambda item_app_key: self.app_get_views(profile=profile, app_key=item_app_key, app_keys=None), ) normalized_args = {"app_key": app_key} return _safe_tool_call( lambda: self._facade.app_get_views(profile=profile, app_key=app_key), error_code="APP_GET_VIEWS_FAILED", normalized_args=normalized_args, suggested_next_call={"tool_name": "app_get_views", "arguments": {"profile": profile, "app_key": app_key}}, ) def app_read_flow_summary(self, *, profile: str, app_key: str) -> JSONObject: """执行应用相关逻辑。""" normalized_args = {"app_key": app_key} return _safe_tool_call( lambda: self._facade.app_read_flow_summary(profile=profile, app_key=app_key), error_code="FLOW_READ_FAILED", normalized_args=normalized_args, suggested_next_call={"tool_name": "app_get_flow", "arguments": {"profile": profile, "app_key": app_key}}, ) def app_get_flow(self, *, profile: str, app_key: str, app_keys: list[str] | None = None) -> JSONObject: """执行应用相关逻辑。""" if app_keys is not None: return self._read_app_batch( tool_name="app_get_flow", profile=profile, app_keys=app_keys, read_one=lambda item_app_key: self.app_get_flow(profile=profile, app_key=item_app_key, app_keys=None), ) normalized_args = {"app_key": app_key} return _safe_tool_call( lambda: self._facade.app_get_flow(profile=profile, app_key=app_key), error_code="APP_GET_FLOW_FAILED", normalized_args=normalized_args, suggested_next_call={"tool_name": "app_get_flow", "arguments": {"profile": profile, "app_key": app_key}}, ) def app_flow_get(self, *, profile: str, app_key: str, version_id: str | None = None) -> JSONObject: normalized_args = {"app_key": app_key, "version_id": version_id} return _safe_tool_call( lambda: self._facade.flow_get(profile=profile, app_key=app_key, version_id=version_id), error_code="APP_FLOW_GET_FAILED", normalized_args=normalized_args, suggested_next_call={"tool_name": "app_flow_get", "arguments": {"profile": profile, **normalized_args}}, ) def app_flow_get_schema(self, *, profile: str, schema_version: str | None = None) -> JSONObject: normalized_args = {"schema_version": schema_version} return _safe_tool_call( lambda: self._facade.flow_get_schema(profile=profile, schema_version=schema_version), error_code="APP_FLOW_SCHEMA_GET_FAILED", normalized_args=normalized_args, suggested_next_call={"tool_name": "app_flow_get_schema", "arguments": {"profile": profile, **normalized_args}}, ) def app_read_charts_summary(self, *, profile: str, app_key: str) -> JSONObject: """执行应用相关逻辑。""" normalized_args = {"app_key": app_key} return _safe_tool_call( lambda: self._facade.app_read_charts_summary(profile=profile, app_key=app_key), error_code="CHARTS_READ_FAILED", normalized_args=normalized_args, suggested_next_call={"tool_name": "app_get_charts", "arguments": {"profile": profile, "app_key": app_key}}, ) def app_get_charts(self, *, profile: str, app_key: str, app_keys: list[str] | None = None) -> JSONObject: """执行应用相关逻辑。""" if app_keys is not None: return self._read_app_batch( tool_name="app_get_charts", profile=profile, app_keys=app_keys, read_one=lambda item_app_key: self.app_get_charts(profile=profile, app_key=item_app_key, app_keys=None), ) normalized_args = {"app_key": app_key} return _safe_tool_call( lambda: self._facade.app_get_charts(profile=profile, app_key=app_key), error_code="APP_GET_CHARTS_FAILED", normalized_args=normalized_args, suggested_next_call={"tool_name": "app_get_charts", "arguments": {"profile": profile, "app_key": app_key}}, ) def app_get_buttons(self, *, profile: str, app_key: str = "", app_keys: list[str] | None = None) -> JSONObject: if app_keys is not None: return self._read_app_batch( tool_name="app_get_buttons", profile=profile, app_keys=app_keys, read_one=lambda item_app_key: self.app_get_buttons(profile=profile, app_key=item_app_key, app_keys=None), ) normalized_args = {"app_key": app_key} return _safe_tool_call( lambda: self._facade.app_get_buttons(profile=profile, app_key=app_key), error_code="APP_GET_BUTTONS_FAILED", normalized_args=normalized_args, suggested_next_call={"tool_name": "app_get_buttons", "arguments": {"profile": profile, "app_key": app_key}}, ) def app_get_associated_resources(self, *, profile: str, app_key: str = "", app_keys: list[str] | None = None) -> JSONObject: if app_keys is not None: return self._read_app_batch( tool_name="app_get_associated_resources", profile=profile, app_keys=app_keys, read_one=lambda item_app_key: self.app_get_associated_resources(profile=profile, app_key=item_app_key, app_keys=None), ) normalized_args = {"app_key": app_key} return _safe_tool_call( lambda: self._facade.app_get_associated_resources(profile=profile, app_key=app_key), error_code="APP_GET_ASSOCIATED_RESOURCES_FAILED", normalized_args=normalized_args, suggested_next_call={"tool_name": "app_get_associated_resources", "arguments": {"profile": profile, "app_key": app_key}}, ) def portal_list(self, *, profile: str) -> JSONObject: """执行门户相关逻辑。""" return _publicize_package_fields(_safe_tool_call( lambda: self._facade.portal_list(profile=profile), error_code="PORTAL_LIST_FAILED", normalized_args={}, suggested_next_call={"tool_name": "portal_list", "arguments": {"profile": profile}}, )) def portal_get(self, *, profile: str, dash_key: str, being_draft: bool = True) -> JSONObject: """执行门户相关逻辑。""" normalized_args = {"dash_key": dash_key, "being_draft": being_draft} return _publicize_package_fields(_safe_tool_call( lambda: self._facade.portal_get(profile=profile, dash_key=dash_key, being_draft=being_draft), error_code="PORTAL_GET_FAILED", normalized_args=normalized_args, suggested_next_call={"tool_name": "portal_get", "arguments": {"profile": profile, **normalized_args}}, )) def portal_read_summary(self, *, profile: str, dash_key: str, being_draft: bool = True) -> JSONObject: """执行门户相关逻辑。""" normalized_args = {"dash_key": dash_key, "being_draft": being_draft} return _publicize_package_fields(_safe_tool_call( lambda: self._facade.portal_read_summary(profile=profile, dash_key=dash_key, being_draft=being_draft), error_code="PORTAL_READ_FAILED", normalized_args=normalized_args, suggested_next_call={"tool_name": "portal_get", "arguments": {"profile": profile, **normalized_args}}, )) def view_get(self, *, profile: str, view_key: str = "", viewgraph_key: str = "") -> JSONObject: """执行视图相关逻辑。""" resolved_view_key = _normalize_builder_view_key(str(view_key or viewgraph_key or "").strip()) normalized_args = {"view_key": resolved_view_key} return _safe_tool_call( lambda: self._facade.view_get(profile=profile, view_key=resolved_view_key), error_code="VIEW_GET_FAILED", normalized_args=normalized_args, suggested_next_call={"tool_name": "view_get", "arguments": {"profile": profile, **normalized_args}}, ) def chart_get( self, *, profile: str, chart_id: str, ) -> JSONObject: """执行图表相关逻辑。""" normalized_args = {"chart_id": chart_id} return _safe_tool_call( lambda: self._facade.chart_get(profile=profile, chart_id=chart_id), error_code="CHART_GET_FAILED", normalized_args=normalized_args, suggested_next_call={"tool_name": "chart_get", "arguments": {"profile": profile, "chart_id": chart_id}}, ) def app_schema_plan( self, *, profile: str, app_key: str = "", package_id: int | None = None, app_name: str = "", icon: str = "", color: str = "", visibility: JSONObject | None = None, add_fields: list[JSONObject], update_fields: list[JSONObject], remove_fields: list[JSONObject], ) -> JSONObject: """执行应用相关逻辑。""" system_field_failure = _reserved_system_field_name_failure( tool_name="app_schema_plan", profile=profile, app_key=app_key, package_id=package_id, app_name=app_name, icon=icon, color=color, visibility=visibility, add_fields=add_fields, update_fields=update_fields, remove_fields=remove_fields, ) if system_field_failure is not None: return system_field_failure try: request = SchemaPlanRequest.model_validate( { "app_key": app_key, "package_tag_id": package_id, "app_name": app_name, "icon": icon, "color": color, "visibility": visibility, "add_fields": add_fields, "update_fields": update_fields, "remove_fields": remove_fields, } ) except ValidationError as exc: return _visibility_validation_failure( str(exc), tool_name="app_schema_plan", exc=exc, suggested_next_call={ "tool_name": "app_schema_plan", "arguments": { "profile": profile, "app_key": app_key, "package_id": package_id, "app_name": app_name, "icon": icon, "color": color, "visibility": visibility, "add_fields": [{"name": "字段名称", "type": "text"}], "update_fields": [], "remove_fields": [], }, }, ) normalized_args = _publicize_package_fields(request.model_dump(mode="json")) return _publicize_package_fields(_safe_tool_call( lambda: self._facade.app_schema_plan(profile=profile, request=request), error_code="SCHEMA_PLAN_FAILED", normalized_args=normalized_args, suggested_next_call={"tool_name": "app_schema_plan", "arguments": {"profile": profile, **normalized_args}}, )) def app_layout_plan( self, *, profile: str, app_key: str, mode: str = "merge", sections: list[JSONObject] | None = None, preset: str | None = None, ) -> JSONObject: """执行应用相关逻辑。""" try: request = LayoutPlanRequest.model_validate( { "app_key": app_key, "mode": mode, "sections": sections or [], "preset": preset, } ) except ValidationError as exc: return _validation_failure( str(exc), tool_name="app_layout_plan", exc=exc, suggested_next_call={ "tool_name": "app_layout_plan", "arguments": { "profile": profile, "app_key": app_key, "mode": "merge", "sections": [{ "type": "paragraph", "paragraph_id": "basic", "title": "基础信息", "rows": [["字段A", "字段B", "字段C", "字段D"]], }], }, }, ) normalized_request = request.model_dump(mode="json", exclude_none=True) return _safe_tool_call( lambda: self._facade.app_layout_plan(profile=profile, request=request), error_code="LAYOUT_PLAN_FAILED", normalized_args=normalized_request, suggested_next_call={"tool_name": "app_layout_plan", "arguments": {"profile": profile, **normalized_request}}, ) def app_flow_plan( self, *, profile: str, app_key: str, mode: str = "replace", nodes: list[JSONObject] | None = None, transitions: list[JSONObject] | None = None, preset: str | None = None, ) -> JSONObject: """执行应用相关逻辑。""" try: request = FlowPlanRequest.model_validate( { "app_key": app_key, "mode": mode, "nodes": nodes or [], "transitions": transitions or [], "preset": preset, } ) except ValidationError as exc: return _validation_failure( str(exc), tool_name="app_flow_plan", exc=exc, suggested_next_call={ "tool_name": "app_flow_plan", "arguments": { "profile": profile, "app_key": app_key, "mode": "replace", "preset": "basic_approval", "nodes": [], "transitions": [], }, }, ) return _safe_tool_call( lambda: self._facade.app_flow_plan(profile=profile, request=request), error_code="FLOW_PLAN_FAILED", normalized_args=request.model_dump(mode="json"), suggested_next_call={"tool_name": "app_flow_plan", "arguments": {"profile": profile, **request.model_dump(mode="json")}}, ) def app_views_plan( self, *, profile: str, app_key: str, upsert_views: list[JSONObject] | None = None, patch_views: list[JSONObject] | None = None, remove_views: list[str] | None = None, preset: str | None = None, ) -> JSONObject: """执行应用相关逻辑。""" reserved_failure = _reserved_system_view_name_failure( tool_name="app_views_plan", profile=profile, app_key=app_key, upsert_views=upsert_views, remove_views=remove_views, ) if reserved_failure is not None: return reserved_failure try: request = ViewsPlanRequest.model_validate( { "app_key": app_key, "upsert_views": upsert_views or [], "patch_views": patch_views or [], "remove_views": remove_views or [], "preset": preset, } ) except ValidationError as exc: return _visibility_validation_failure( str(exc), tool_name="app_views_plan", exc=exc, suggested_next_call={ "tool_name": "app_views_plan", "arguments": { "profile": profile, "app_key": app_key, "preset": "default_table", "upsert_views": [], "patch_views": [], "remove_views": [], }, }, ) return _safe_tool_call( lambda: self._facade.app_views_plan(profile=profile, request=request), error_code="VIEWS_PLAN_FAILED", normalized_args={ "app_key": request.app_key, "upsert_views": [public_view_upsert_payload(view) for view in request.upsert_views], "patch_views": [public_view_partial_payload(patch) for patch in request.patch_views], "remove_views": list(request.remove_views), "preset": request.preset.value if request.preset is not None else None, }, suggested_next_call={ "tool_name": "app_views_plan", "arguments": { "profile": profile, "app_key": request.app_key, "upsert_views": [public_view_upsert_payload(view) for view in request.upsert_views], "patch_views": [public_view_partial_payload(patch) for patch in request.patch_views], "remove_views": list(request.remove_views), "preset": request.preset.value if request.preset is not None else None, }, }, ) def app_schema_apply( self, *, profile: str, app_key: str = "", package_id: int | None = None, app_name: str = "", app_title: str = "", icon: str | JSONObject = "", color: str = "", icon_name: str | None = None, icon_color: str | None = None, icon_config: JSONObject | None = None, visibility: JSONObject | None = None, publish: bool = True, form: list[JSONObject] | None = None, add_fields: list[JSONObject], update_fields: list[JSONObject], remove_fields: list[JSONObject], apps: list[JSONObject] | None = None, ) -> JSONObject: """执行应用相关逻辑。""" icon, color = _normalize_builder_icon_args( icon=icon, color=color, icon_name=icon_name, icon_color=icon_color, icon_config=icon_config, ) if apps is not None: normalized_apps_payload = _normalize_schema_apps_argument( tool_name="app_schema_apply", package_id=package_id, apps=apps, ) failure = normalized_apps_payload.get("failure") if isinstance(failure, dict): return _attach_builder_apply_envelope("app_schema_apply", failure) package_id = normalized_apps_payload.get("package_id") # type: ignore[assignment] apps = normalized_apps_payload.get("apps") # type: ignore[assignment] input_warnings = list(normalized_apps_payload.get("warnings") or []) normalized_apps: list[JSONObject] = [] for index, app_item in enumerate(apps or []): if not isinstance(app_item, dict): normalized_apps.append(app_item) continue form_result = _apply_schema_form_to_payload( tool_name="app_schema_apply", payload=app_item, path=f"apps[{index}]", ) failure = form_result.get("failure") if isinstance(failure, dict): return _attach_builder_apply_envelope("app_schema_apply", failure) normalized_apps.append(form_result.get("payload") or app_item) apps = normalized_apps result = self._app_schema_apply_multi( profile=profile, package_id=package_id, visibility=visibility, publish=publish, apps=apps, ) if input_warnings: result_warnings = list(result.get("warnings") or []) result_warnings.extend(deepcopy(input_warnings)) result["warnings"] = result_warnings return _attach_builder_apply_envelope("app_schema_apply", result) inline_form_layout_sections: list[JSONObject] = [] if form: form_payload: JSONObject = {"form": form} if add_fields: form_payload["add_fields"] = add_fields form_result = _apply_schema_form_to_payload( tool_name="app_schema_apply", payload=form_payload, path="schema", ) failure = form_result.get("failure") if isinstance(failure, dict): return _attach_builder_apply_envelope("app_schema_apply", failure) normalized_payload = form_result.get("payload") if isinstance(form_result.get("payload"), dict) else {} add_fields = list(normalized_payload.get("add_fields") or []) inline_form_layout_sections = list(form_result.get("layout_sections") or []) result = self._app_schema_apply_once( profile=profile, app_key=app_key, package_id=package_id, app_name=app_name, app_title=app_title, icon=icon, color=color, visibility=visibility, publish=publish, add_fields=add_fields, update_fields=update_fields, remove_fields=remove_fields, inline_form_layout_sections=inline_form_layout_sections, ) result = self._retry_after_self_lock_release( profile=profile, result=result, retry_call=lambda: self._app_schema_apply_once( profile=profile, app_key=app_key, package_id=package_id, app_name=app_name, app_title=app_title, icon=icon, color=color, visibility=visibility, publish=publish, add_fields=add_fields, update_fields=update_fields, remove_fields=remove_fields, inline_form_layout_sections=inline_form_layout_sections, ), ) return _attach_builder_apply_envelope("app_schema_apply", result) def _app_schema_apply_multi( self, *, profile: str, package_id: int | None, visibility: JSONObject | None, publish: bool, apps: list[JSONObject], ) -> JSONObject: started_at = time.perf_counter() validation_ms = 0 app_shell_apply_ms = 0 relation_compile_ms = 0 app_field_apply_ms = 0 relation_repair_ms = 0 inline_layout_fallback_ms = 0 child_schema_write_ms = 0 child_inline_layout_ms = 0 child_publish_ms = 0 child_readback_ms = 0 normalized_args: JSONObject = { "package_id": package_id, "publish": publish, "apps": deepcopy(apps), } if visibility is not None: normalized_args["visibility"] = deepcopy(visibility) parallelism: JSONObject = {"shell_workers": 0, "schema_workers": 0} def merge_child_duration(child: JSONObject) -> None: nonlocal child_schema_write_ms nonlocal child_inline_layout_ms nonlocal child_publish_ms nonlocal child_readback_ms breakdown = child.get("duration_breakdown_ms") if isinstance(child, dict) else None if not isinstance(breakdown, dict): return for key, target in ( ("schema_write_ms", "child_schema_write_ms"), ("inline_layout_ms", "child_inline_layout_ms"), ("publish_ms", "child_publish_ms"), ("readback_ms", "child_readback_ms"), ): value = breakdown.get(key) if isinstance(value, (int, float)) and value >= 0: if target == "child_schema_write_ms": child_schema_write_ms += int(value) elif target == "child_inline_layout_ms": child_inline_layout_ms += int(value) elif target == "child_publish_ms": child_publish_ms += int(value) elif target == "child_readback_ms": child_readback_ms += int(value) def finish(response: JSONObject) -> JSONObject: _merge_duration_breakdown( response, multi_app_validation_ms=validation_ms, app_shell_apply_ms=app_shell_apply_ms, app_shell_parallel_wall_ms=app_shell_apply_ms, relation_compile_ms=relation_compile_ms, app_field_apply_ms=app_field_apply_ms, app_schema_parallel_wall_ms=app_field_apply_ms, relation_repair_parallel_wall_ms=relation_repair_ms, native_inline_layout_ms=child_inline_layout_ms, inline_layout_fallback_ms=inline_layout_fallback_ms, child_schema_write_ms=child_schema_write_ms, child_inline_layout_ms=child_inline_layout_ms, child_publish_ms=child_publish_ms, child_readback_ms=child_readback_ms, total_ms=_elapsed_ms(started_at), ) return response validation_started_at = time.perf_counter() system_field_failure = _reserved_system_field_name_failure_for_apps( tool_name="app_schema_apply", profile=profile, package_id=package_id, visibility=visibility, publish=publish, apps=apps, ) if system_field_failure is not None: validation_ms += _elapsed_ms(validation_started_at) return finish(system_field_failure) if package_id is None: validation_ms += _elapsed_ms(validation_started_at) return finish( _config_failure( tool_name="app_schema_apply", message="app_schema_apply multi-app mode requires package_id.", fix_hint="Pass package_id and apps[].app_name for new apps, or apps[].app_key for existing apps.", ) ) if not apps: validation_ms += _elapsed_ms(validation_started_at) return finish( _config_failure( tool_name="app_schema_apply", error_code="APPS_FILE_EMPTY", message="app_schema_apply multi-app mode requires non-empty apps.", fix_hint="Pass apps as a non-empty list of app schema items.", ) ) shape_failure = _schema_apps_shape_failure(tool_name="app_schema_apply", apps=apps) if shape_failure is not None: validation_ms += _elapsed_ms(validation_started_at) return finish(shape_failure) static_validation_failure = _multi_app_static_validation_failure( tool_name="app_schema_apply", apps=apps, ) if static_validation_failure is not None: validation_ms += _elapsed_ms(validation_started_at) return finish(static_validation_failure) icon_errors: list[JSONObject] = [] seen_new_app_icons: dict[str, int] = {} for index, raw_item in enumerate(apps): if not isinstance(raw_item, dict): continue app_key = str(raw_item.get("app_key") or raw_item.get("appKey") or "").strip() creating_item = not app_key icon_failure = _validate_workspace_icon_for_builder( tool_name="app_schema_apply", icon=str(raw_item.get("icon") or ""), color=str(raw_item.get("color") or ""), creating=creating_item, ) if icon_failure is not None: icon_errors.append( { "index": index, "row_number": index + 1, "error_code": icon_failure.get("error_code"), "message": icon_failure.get("message"), "details": icon_failure.get("details"), } ) continue _ok, _error_code, _message, icon_details = validate_workspace_icon_choice( icon=str(raw_item.get("icon") or ""), color=str(raw_item.get("color") or ""), require_explicit=creating_item, disallow_generic=creating_item, ) normalized_icon = str(icon_details.get("normalized_icon") or "").strip() if creating_item and normalized_icon: if normalized_icon in seen_new_app_icons: icon_errors.append( { "index": index, "row_number": index + 1, "error_code": "DUPLICATE_WORKSPACE_ICON_IN_BATCH", "message": f"apps[{index}] reuses icon '{normalized_icon}' from apps[{seen_new_app_icons[normalized_icon]}]", "details": { "icon": normalized_icon, "first_index": seen_new_app_icons[normalized_icon], "duplicate_index": index, "icon_catalog_command": "qingflow --json builder icon catalog", }, } ) else: seen_new_app_icons[normalized_icon] = index if icon_errors: validation_ms += _elapsed_ms(validation_started_at) return finish( _config_failure( tool_name="app_schema_apply", error_code="WORKSPACE_ICON_BATCH_INVALID", message="one or more apps have invalid workspace icon configuration", fix_hint="Call `qingflow --json builder icon catalog`, choose a distinct non-template icon and color for each new app, then retry.", details={"icon_errors": icon_errors}, allowed_values={ "workspace_icon.icon_names": list(WORKSPACE_ICON_NAMES), "workspace_icon.icon_colors": list(WORKSPACE_ICON_COLORS), "workspace_icon.generic_icon_names": list(GENERIC_WORKSPACE_ICON_NAMES), }, ) ) validation_ms += _elapsed_ms(validation_started_at) client_key_to_app_key: dict[str, str] = {} app_name_to_app_key: dict[str, str] = {} created_app_keys: list[str] = [] results_by_index: list[JSONObject | None] = [None] * len(apps) shell_jobs: list[JSONObject] = [] any_write_executed = False any_write_may_have_succeeded = False pending_readback_app_keys: list[str] = [] pending_readback_app_names: list[str] = [] client_keys: set[str] = set() client_key_indexes: dict[str, int] = {} app_name_indexes: dict[str, int] = {} def worker_count(count: int) -> int: return min(MULTI_APP_SCHEMA_APPLY_PARALLEL_LIMIT, count) if count > 0 else 0 def mark_inline_layout_not_applied_in_schema( result: JSONObject, *, app_key: str, sections: list[JSONObject], ) -> JSONObject: marked = deepcopy(result) inline_payload: JSONObject = { "requested": True, "section_count": len(sections), "app_key": app_key or None, "applied": False, "applied_in_schema": False, "error_code": "INLINE_FORM_LAYOUT_NOT_APPLIED_IN_SCHEMA", "message": "schema save did not apply inline form layout; multi-app mode does not run app_layout_apply fallback", } marked["inline_form_layout"] = inline_payload verification = marked.get("verification") if isinstance(marked.get("verification"), dict) else {} verification = deepcopy(verification) verification["inline_form_layout_verified"] = False marked["verification"] = verification if marked.get("status") in {"success", "partial_success"}: marked["status"] = "partial_success" marked["error_code"] = marked.get("error_code") or "INLINE_FORM_LAYOUT_NOT_APPLIED_IN_SCHEMA" marked["message"] = "schema applied but inline form layout was not applied in schema save" marked["safe_to_retry"] = False return marked def dependency_failure_for_item(index: int, item: JSONObject) -> JSONObject | None: blocked: list[JSONObject] = [] for ref in _collect_multi_app_target_refs(item, path=f"apps[{index}]"): ref_value = str(ref.get("value") or "").strip() if not ref_value: continue target_index: int | None = None resolved = False if ref.get("kind") == "target_app_ref": target_index = client_key_indexes.get(ref_value) resolved = ref_value in client_key_to_app_key elif ref.get("kind") == "target_app": target_index = app_name_indexes.get(ref_value) resolved = ref_value in app_name_to_app_key if target_index is None or resolved: continue target_result = results_by_index[target_index] if 0 <= target_index < len(results_by_index) else None blocked.append( { "kind": ref.get("kind"), "value": ref_value, "path": ref.get("path"), "target_index": target_index, "target_status": target_result.get("status") if isinstance(target_result, dict) else None, "target_error_code": target_result.get("error_code") if isinstance(target_result, dict) else None, } ) if not blocked: return None raw_item = apps[index] if 0 <= index < len(apps) else item failure = _multi_app_item_failure( index, raw_item, "DEPENDENCY_APP_NOT_READY", "one or more target apps failed or require readback, so this app schema was not written", ) failure["stage"] = "dependency_check" failure["safe_to_retry"] = False failure["details"] = {"blocked_dependencies": blocked} return failure for index, raw_item in enumerate(apps): if not isinstance(raw_item, dict): results_by_index[index] = _multi_app_item_failure(index, raw_item, "INVALID_APP_ITEM", "apps[] items must be objects") continue item = deepcopy(raw_item) client_key = str(item.get("client_key") or item.get("clientKey") or "").strip() app_name = str(item.get("app_name") or item.get("appTitle") or item.get("app_title") or item.get("title") or "").strip() app_key = str(item.get("app_key") or item.get("appKey") or "").strip() if client_key: if client_key in client_keys: results_by_index[index] = _multi_app_item_failure(index, item, "DUPLICATE_CLIENT_KEY", f"duplicate client_key '{client_key}'") continue client_keys.add(client_key) client_key_indexes[client_key] = index if app_name: app_name_indexes[app_name] = index if not app_key and not app_name: results_by_index[index] = _multi_app_item_failure(index, item, "APP_SELECTOR_REQUIRED", "apps[] requires app_key or app_name") continue shell_jobs.append( { "index": index, "item": item, "client_key": client_key, "app_name": app_name, "app_key": app_key, } ) shell_workers = worker_count(len(shell_jobs)) parallelism["shell_workers"] = shell_workers def run_shell_job(job: JSONObject) -> JSONObject: index = int(job["index"]) item = deepcopy(job["item"]) app_key = str(job.get("app_key") or "") app_name = str(job.get("app_name") or "") shell_started_at = time.perf_counter() try: shell = self._app_schema_apply_once( profile=profile, app_key=app_key, package_id=package_id if not app_key else None, app_name=app_name, app_title="", icon=str(item.get("icon") or ""), color=str(item.get("color") or ""), visibility=item.get("visibility", visibility), publish=False, add_fields=[], update_fields=[], remove_fields=[], inline_form_layout_sections=None, allow_inline_layout_fallback=False, ) return { "index": index, "job": job, "elapsed_ms": _elapsed_ms(shell_started_at), "public_shell": _publicize_package_fields(shell), } except Exception as error: # pragma: no cover - defensive guard for unexpected worker failures return { "index": index, "job": job, "elapsed_ms": _elapsed_ms(shell_started_at), "exception": repr(error), } shell_outputs: list[JSONObject] = [] shell_phase_started_at = time.perf_counter() if shell_jobs: with ThreadPoolExecutor(max_workers=shell_workers) as executor: futures = [executor.submit(run_shell_job, job) for job in shell_jobs] for future in as_completed(futures): shell_outputs.append(future.result()) app_shell_apply_ms += _elapsed_ms(shell_phase_started_at) for output in sorted(shell_outputs, key=lambda item: int(item.get("index", 0))): index = int(output["index"]) job = output.get("job") if isinstance(output.get("job"), dict) else {} client_key = str(job.get("client_key") or "").strip() app_name = str(job.get("app_name") or "").strip() app_key = str(job.get("app_key") or "").strip() item = deepcopy(job.get("item") if isinstance(job.get("item"), dict) else {}) exception = output.get("exception") if exception: results_by_index[index] = { "index": index, "row_number": index + 1, "client_key": client_key or None, "app_name": app_name or None, "app_key": app_key or None, "status": "pending_readback", "stage": "resolve_or_create_shell", "error_code": "APP_SHELL_APPLY_EXCEPTION", "message": str(exception), "write_executed": False, "write_may_have_succeeded": True, "safe_to_retry": False, "next_action": "readback_before_retry", } if app_name: pending_readback_app_names.append(app_name) any_write_may_have_succeeded = True continue public_shell = output.get("public_shell") if isinstance(output.get("public_shell"), dict) else {} merge_child_duration(public_shell) resolved_key = str(public_shell.get("app_key") or "").strip() shell_write_executed = _schema_apply_result_has_write(public_shell) shell_write_may_have_succeeded = _schema_apply_result_may_have_write(public_shell) if shell_write_executed: any_write_executed = True if shell_write_may_have_succeeded: any_write_may_have_succeeded = True if public_shell.get("status") not in {"success", "partial_success"} or not resolved_key: pending_readback = bool(shell_write_may_have_succeeded or shell_write_executed) and ( str(public_shell.get("next_action") or "") == "readback_before_retry" or bool((public_shell.get("verification") if isinstance(public_shell.get("verification"), dict) else {}).get("readback_before_retry")) or not resolved_key ) item_status = "pending_readback" if pending_readback else "failed" if pending_readback: if resolved_key: pending_readback_app_keys.append(resolved_key) if app_name: pending_readback_app_names.append(app_name) results_by_index[index] = { "index": index, "row_number": index + 1, "client_key": client_key or None, "app_name": app_name or None, "app_key": resolved_key or app_key or None, "status": item_status, "stage": "resolve_or_create_shell", "error_code": public_shell.get("error_code") or "APP_SHELL_APPLY_FAILED", "message": public_shell.get("message") or "app shell resolve/create failed", "write_executed": shell_write_executed, "write_may_have_succeeded": bool(shell_write_may_have_succeeded or shell_write_executed), "safe_to_retry": False if pending_readback else (not shell_write_executed and not any_write_executed), **({"next_action": "readback_before_retry"} if pending_readback else {}), **({"verification": public_shell.get("verification")} if isinstance(public_shell.get("verification"), dict) else {}), **({"created": True} if pending_readback and not app_key else {}), } continue if bool(public_shell.get("created")): created_app_keys.append(resolved_key) if shell_write_may_have_succeeded and str(public_shell.get("next_action") or "") == "readback_before_retry": pending_readback_app_keys.append(resolved_key) if app_name: pending_readback_app_names.append(app_name) if client_key: client_key_to_app_key[client_key] = resolved_key resolved_name = str(public_shell.get("app_name_after") or public_shell.get("app_name") or app_name or "").strip() if resolved_name: app_name_to_app_key[resolved_name] = resolved_key results_by_index[index] = { "index": index, "row_number": index + 1, "client_key": client_key or None, "app_name": resolved_name or None, "app_key": resolved_key, "status": "shell_ready", "created": bool(public_shell.get("created")), "shell_result": public_shell, "shell_field_diff": public_shell.get("field_diff") or {}, "shell_field_diff_details": public_shell.get("field_diff_details") or {}, } final_items_by_index: list[JSONObject | None] = [None] * len(apps) schema_jobs: list[JSONObject] = [] for index, raw_item in enumerate(apps): existing = results_by_index[index] if 0 <= index < len(results_by_index) else None if not existing or existing.get("status") != "shell_ready": if existing: final_items_by_index[index] = existing continue item = deepcopy(raw_item) app_key = str(existing.get("app_key") or "").strip() dependency_failure = dependency_failure_for_item(index, item) if dependency_failure is not None: final_items_by_index[index] = dependency_failure any_write_executed = True continue compile_started_at = time.perf_counter() try: compiled_item = _compile_multi_app_schema_item_refs(item, client_key_to_app_key, app_name_to_app_key) except ValueError as error: relation_compile_ms += _elapsed_ms(compile_started_at) final_items_by_index[index] = { **{key: existing.get(key) for key in ("index", "row_number", "client_key", "app_name", "app_key", "created")}, "status": "failed", "stage": "compile_relation_refs", "error_code": "TARGET_APP_REF_NOT_FOUND", "message": str(error), "safe_to_retry": False, } any_write_executed = True continue relation_compile_ms += _elapsed_ms(compile_started_at) schema_add_fields = list(compiled_item.get("add_fields") or []) update_fields = list(compiled_item.get("update_fields") or []) remove_fields = list(compiled_item.get("remove_fields") or []) layout_sections = list(compiled_item.get("_inline_form_layout_sections") or []) field_phase_app_name = "" field_phase_icon = "" field_phase_color = "" field_phase_visibility = None relation_fields_for_retry = [ deepcopy(field) for field in schema_add_fields if isinstance(field, dict) and _multi_app_field_type(field) == PublicFieldType.relation.value ] schema_jobs.append( { "index": index, "existing": existing, "item": item, "app_key": app_key, "field_phase_app_name": field_phase_app_name, "field_phase_icon": field_phase_icon, "field_phase_color": field_phase_color, "field_phase_visibility": field_phase_visibility, "schema_add_fields": schema_add_fields, "update_fields": update_fields, "remove_fields": remove_fields, "layout_sections": layout_sections, "relation_fields_for_retry": relation_fields_for_retry, } ) schema_workers = worker_count(len(schema_jobs)) parallelism["schema_workers"] = schema_workers parallelism["schema_wave_count"] = 0 parallelism["schema_wave_sizes"] = [] def run_schema_job(job: JSONObject) -> JSONObject: index = int(job["index"]) app_key = str(job.get("app_key") or "") field_phase_app_name = str(job.get("field_phase_app_name") or "") field_phase_icon = str(job.get("field_phase_icon") or "") field_phase_color = str(job.get("field_phase_color") or "") field_phase_visibility = job.get("field_phase_visibility") schema_add_fields = list(job.get("schema_add_fields") or []) update_fields = list(job.get("update_fields") or []) remove_fields = list(job.get("remove_fields") or []) layout_sections = list(job.get("layout_sections") or []) relation_fields_for_retry = list(job.get("relation_fields_for_retry") or []) field_started_at = time.perf_counter() try: field_result = self._app_schema_apply_once( profile=profile, app_key=app_key, package_id=None, app_name=field_phase_app_name, app_title="", icon=field_phase_icon, color=field_phase_color, visibility=field_phase_visibility, publish=publish, add_fields=schema_add_fields, update_fields=update_fields, remove_fields=remove_fields, inline_form_layout_sections=layout_sections, allow_inline_layout_fallback=False, ) public_result = _publicize_package_fields(field_result) except Exception as error: # pragma: no cover - defensive guard for unexpected worker failures return { "index": index, "job": job, "elapsed_ms": _elapsed_ms(field_started_at), "exception": repr(error), } if _multi_app_deferred_relation_retryable(public_result, relation_fields_for_retry): first_error = { "error_code": public_result.get("error_code"), "message": public_result.get("message"), "backend_code": public_result.get("backend_code"), "http_status": public_result.get("http_status"), "details": deepcopy(public_result.get("details")) if isinstance(public_result.get("details"), dict) else None, } retry_attempts: list[JSONObject] = [] for retry_index, delay_seconds in enumerate((1.0, 5.0), start=1): retry_readback = self._stabilize_deferred_relation_metadata( profile=profile, source_app_key=app_key, deferred_add_fields=relation_fields_for_retry, delay_seconds=delay_seconds, ) retry_result = self._app_schema_apply_once( profile=profile, app_key=app_key, package_id=None, app_name=field_phase_app_name, app_title="", icon=field_phase_icon, color=field_phase_color, visibility=field_phase_visibility, publish=publish, add_fields=schema_add_fields, update_fields=update_fields, remove_fields=remove_fields, inline_form_layout_sections=layout_sections, allow_inline_layout_fallback=False, ) public_retry_result = _publicize_package_fields(retry_result) retry_attempts.append( { "attempt": retry_index, "delay_seconds": delay_seconds, "status": public_retry_result.get("status"), "error_code": public_retry_result.get("error_code"), "backend_code": public_retry_result.get("backend_code"), "http_status": public_retry_result.get("http_status"), "readback": retry_readback, } ) public_result = public_retry_result if not _multi_app_deferred_relation_retryable(public_result, relation_fields_for_retry): break retry_details = public_result.get("details") if not isinstance(retry_details, dict): retry_details = {} public_result["details"] = retry_details retry_summary: JSONObject = { "attempted": True, "reason": "same_batch_relation_metadata_may_lag", "first_error": first_error, "attempts": retry_attempts, } if retry_attempts: retry_summary["readback"] = retry_attempts[-1].get("readback") retry_details["deferred_relation_retry"] = retry_summary if layout_sections and not _inline_form_layout_was_handled(public_result): public_result = mark_inline_layout_not_applied_in_schema( public_result, app_key=app_key, sections=layout_sections, ) return { "index": index, "job": job, "elapsed_ms": _elapsed_ms(field_started_at), "public_result": public_result, } def schema_job_dependency_indexes(job: JSONObject) -> set[int]: index = int(job.get("index", -1)) item = job.get("item") if isinstance(job.get("item"), dict) else {} dependencies: set[int] = set() for ref in _collect_multi_app_target_refs(item, path=f"apps[{index}]"): ref_value = str(ref.get("value") or "").strip() if not ref_value: continue target_index: int | None = None if ref.get("kind") == "target_app_ref": target_index = client_key_indexes.get(ref_value) elif ref.get("kind") == "target_app": target_index = app_name_indexes.get(ref_value) if target_index is None or target_index == index: continue dependencies.add(target_index) return dependencies def schema_dependency_failure(index: int, job: JSONObject, blocked_dependencies: list[JSONObject]) -> JSONObject: existing = job.get("existing") if isinstance(job.get("existing"), dict) else {} return { **{key: existing.get(key) for key in ("index", "row_number", "client_key", "app_name", "app_key", "created")}, "status": "failed", "stage": "schema_dependency_check", "error_code": "DEPENDENCY_SCHEMA_NOT_READY", "message": "one or more same-batch target app schemas failed or are not ready, so this app schema was not written", "write_executed": False, "write_may_have_succeeded": False, "safe_to_retry": False, "details": {"blocked_dependencies": blocked_dependencies}, } def process_schema_output(output: JSONObject) -> None: nonlocal any_write_executed nonlocal any_write_may_have_succeeded index = int(output["index"]) job = output.get("job") if isinstance(output.get("job"), dict) else {} existing = job.get("existing") if isinstance(job.get("existing"), dict) else {} app_key = str(job.get("app_key") or "") exception = output.get("exception") if exception: final_items_by_index[index] = { **{key: existing.get(key) for key in ("index", "row_number", "client_key", "app_name", "app_key", "created")}, "status": "pending_readback", "stage": "schema_apply", "error_code": "SCHEMA_APPLY_EXCEPTION", "message": str(exception), "write_executed": False, "write_may_have_succeeded": True, "safe_to_retry": False, "next_action": "readback_before_retry", } if app_key: pending_readback_app_keys.append(app_key) if existing.get("app_name"): pending_readback_app_names.append(str(existing.get("app_name"))) any_write_may_have_succeeded = True return public_result = output.get("public_result") if isinstance(output.get("public_result"), dict) else {} merge_child_duration(public_result) if _schema_apply_result_has_write(public_result): any_write_executed = True if _schema_apply_result_may_have_write(public_result): any_write_may_have_succeeded = True item_status = public_result.get("status") if public_result.get("status") in {"success", "partial_success"} else "failed" shell_field_diff = existing.get("shell_field_diff") if isinstance(existing.get("shell_field_diff"), dict) else {} shell_field_diff_details = existing.get("shell_field_diff_details") if isinstance(existing.get("shell_field_diff_details"), dict) else {} field_diff = _merge_schema_field_diffs(shell_field_diff, public_result.get("field_diff") or {}) field_diff_details = _merge_schema_field_diffs(shell_field_diff_details, public_result.get("field_diff_details") or {}) if _schema_apply_result_may_have_write(public_result) and str(public_result.get("next_action") or "") == "readback_before_retry": if app_key: pending_readback_app_keys.append(app_key) if existing.get("app_name"): pending_readback_app_names.append(str(existing.get("app_name"))) final_items_by_index[index] = { **{key: existing.get(key) for key in ("index", "row_number", "client_key", "app_name", "app_key", "created")}, "status": item_status, "stage": "schema_apply", "field_diff": field_diff, "field_diff_details": field_diff_details, "shell_field_diff": shell_field_diff, "shell_field_diff_details": shell_field_diff_details, "published": bool(public_result.get("published")), "verified": bool(public_result.get("verified")), "error_code": public_result.get("error_code"), "message": public_result.get("message"), "write_executed": _schema_apply_result_has_write(public_result), "write_may_have_succeeded": _schema_apply_result_may_have_write(public_result), "safe_to_retry": False, **_multi_app_child_diagnostics(public_result), **({"next_action": public_result.get("next_action")} if public_result.get("next_action") else {}), **({"verification": public_result.get("verification")} if isinstance(public_result.get("verification"), dict) else {}), **({"inline_form_layout": public_result.get("inline_form_layout")} if isinstance(public_result.get("inline_form_layout"), dict) else {}), **({"inline_form_layout_result": public_result.get("inline_form_layout_result")} if isinstance(public_result.get("inline_form_layout_result"), dict) else {}), } schema_jobs_by_index = {int(job["index"]): job for job in schema_jobs} schema_dependencies_by_index = { index: schema_job_dependency_indexes(job) for index, job in schema_jobs_by_index.items() } pending_schema_indexes = set(schema_jobs_by_index) schema_phase_started_at = time.perf_counter() while pending_schema_indexes: blocked_this_round = False for index in sorted(list(pending_schema_indexes)): blocked_dependencies: list[JSONObject] = [] for dependency_index in sorted(schema_dependencies_by_index.get(index, set())): if dependency_index not in schema_jobs_by_index: continue dependency_result = final_items_by_index[dependency_index] if dependency_result is None: continue dependency_status = dependency_result.get("status") if dependency_status not in {"success", "partial_success"}: blocked_dependencies.append( { "target_index": dependency_index, "target_status": dependency_status, "target_error_code": dependency_result.get("error_code"), "target_app_key": dependency_result.get("app_key"), "target_app_name": dependency_result.get("app_name"), } ) if not blocked_dependencies: continue final_items_by_index[index] = schema_dependency_failure( index, schema_jobs_by_index[index], blocked_dependencies, ) pending_schema_indexes.remove(index) blocked_this_round = True any_write_executed = True if blocked_this_round: continue ready_indexes = [ index for index in sorted(pending_schema_indexes) if all( dependency_index not in pending_schema_indexes for dependency_index in schema_dependencies_by_index.get(index, set()) if dependency_index in schema_jobs_by_index ) ] if not ready_indexes: for index in sorted(pending_schema_indexes): existing = schema_jobs_by_index[index].get("existing") existing = existing if isinstance(existing, dict) else {} final_items_by_index[index] = { **{key: existing.get(key) for key in ("index", "row_number", "client_key", "app_name", "app_key", "created")}, "status": "failed", "stage": "schema_dependency_check", "error_code": "DEPENDENCY_SCHEMA_CYCLE", "message": "same-batch target_app_ref dependencies contain a cycle; schema was not written", "write_executed": False, "write_may_have_succeeded": False, "safe_to_retry": False, "details": { "dependency_indexes": sorted(schema_dependencies_by_index.get(index, set())), }, } any_write_executed = True pending_schema_indexes.clear() break parallelism["schema_wave_count"] = int(parallelism.get("schema_wave_count") or 0) + 1 schema_wave_sizes = parallelism.get("schema_wave_sizes") if isinstance(schema_wave_sizes, list): schema_wave_sizes.append(len(ready_indexes)) wave_outputs: list[JSONObject] = [] wave_workers = worker_count(len(ready_indexes)) with ThreadPoolExecutor(max_workers=wave_workers) as executor: futures = [executor.submit(run_schema_job, schema_jobs_by_index[index]) for index in ready_indexes] for future in as_completed(futures): wave_outputs.append(future.result()) for output in sorted(wave_outputs, key=lambda item: int(item.get("index", 0))): process_schema_output(output) for index in ready_indexes: pending_schema_indexes.remove(index) app_field_apply_ms += _elapsed_ms(schema_phase_started_at) final_items: list[JSONObject] = [ item if isinstance(item, dict) else _multi_app_item_failure(index, apps[index], "APP_ITEM_NOT_PROCESSED", "app item was not processed") for index, item in enumerate(final_items_by_index) ] def relation_repair_patches(item: JSONObject) -> list[JSONObject]: details = item.get("details") if isinstance(item.get("details"), dict) else {} repair_plan = details.get("relation_repair_plan") if isinstance(details.get("relation_repair_plan"), list) else [] patches: list[JSONObject] = [] for plan_item in repair_plan: if not isinstance(plan_item, dict): continue patch = plan_item.get("suggested_patch") if isinstance(patch, dict): patches.append(deepcopy(patch)) return patches relation_repair_jobs: list[JSONObject] = [] for item in final_items: if item.get("status") != "partial_success": continue app_key = str(item.get("app_key") or "").strip() patches = relation_repair_patches(item) if app_key and patches: relation_repair_jobs.append({"item": item, "app_key": app_key, "patches": patches}) def run_relation_repair_job(job: JSONObject) -> JSONObject: app_key = str(job.get("app_key") or "") patches = list(job.get("patches") or []) repair_started_at = time.perf_counter() try: repair_result = self._app_schema_apply_once( profile=profile, app_key=app_key, package_id=None, app_name="", app_title="", icon="", color="", visibility=None, publish=publish, add_fields=[], update_fields=patches, remove_fields=[], inline_form_layout_sections=None, allow_inline_layout_fallback=False, ) public_repair = _publicize_package_fields(repair_result) except Exception as error: # pragma: no cover - defensive guard for unexpected worker failures return {"job": job, "elapsed_ms": _elapsed_ms(repair_started_at), "exception": repr(error)} return { "job": job, "elapsed_ms": _elapsed_ms(repair_started_at), "public_repair": public_repair, } if relation_repair_jobs: repair_phase_started_at = time.perf_counter() repair_workers = worker_count(len(relation_repair_jobs)) repair_outputs: list[JSONObject] = [] with ThreadPoolExecutor(max_workers=repair_workers) as executor: futures = [executor.submit(run_relation_repair_job, job) for job in relation_repair_jobs] for future in as_completed(futures): repair_outputs.append(future.result()) relation_repair_ms += _elapsed_ms(repair_phase_started_at) for output in repair_outputs: job = output.get("job") if isinstance(output.get("job"), dict) else {} original = job.get("item") if isinstance(job.get("item"), dict) else {} if not original: continue try: index = int(original.get("index")) except (TypeError, ValueError): continue if index < 0 or index >= len(final_items): continue repaired_item = deepcopy(original) exception = output.get("exception") if exception: repaired_item["relation_repair"] = { "attempted": True, "status": "failed", "error_code": "RELATION_REPAIR_EXCEPTION", "message": str(exception), } final_items[index] = repaired_item continue public_repair = output.get("public_repair") if isinstance(output.get("public_repair"), dict) else {} merge_child_duration(public_repair) if _schema_apply_result_has_write(public_repair): any_write_executed = True if _schema_apply_result_may_have_write(public_repair): any_write_may_have_succeeded = True repair_payload: JSONObject = { "attempted": True, "status": public_repair.get("status"), "error_code": public_repair.get("error_code"), "message": public_repair.get("message"), "verified": bool(public_repair.get("verified")), "field_diff": public_repair.get("field_diff") or {}, **({"verification": public_repair.get("verification")} if isinstance(public_repair.get("verification"), dict) else {}), **({"details": public_repair.get("details")} if isinstance(public_repair.get("details"), dict) else {}), **({"duration_breakdown_ms": public_repair.get("duration_breakdown_ms")} if isinstance(public_repair.get("duration_breakdown_ms"), dict) else {}), } repaired_item["relation_repair"] = repair_payload repaired_item["field_diff"] = _merge_schema_field_diffs(repaired_item.get("field_diff") or {}, public_repair.get("field_diff") or {}) repaired_item["field_diff_details"] = _merge_schema_field_diffs( repaired_item.get("field_diff_details") or {}, public_repair.get("field_diff_details") or {}, ) if public_repair.get("status") == "success" and bool(public_repair.get("verified")): repaired_item["status"] = "success" repaired_item["error_code"] = None repaired_item["message"] = "applied schema patch; relation readback repaired" repaired_item["published"] = bool(public_repair.get("published")) repaired_item["verified"] = True original_verification = repaired_item.get("verification") if isinstance(repaired_item.get("verification"), dict) else {} repair_verification = public_repair.get("verification") if isinstance(public_repair.get("verification"), dict) else {} merged_verification = {**deepcopy(original_verification), **deepcopy(repair_verification)} repaired_item["verification"] = merged_verification details = repaired_item.get("details") if isinstance(repaired_item.get("details"), dict) else {} details = deepcopy(details) details.pop("relation_repair_plan", None) repaired_item["details"] = details final_items[index] = repaired_item pending_readback = sum(1 for item in final_items if item.get("status") == "pending_readback") succeeded = sum(1 for item in final_items if item.get("status") in {"success", "partial_success"}) failed = sum(1 for item in final_items if item.get("status") == "failed") partial = sum(1 for item in final_items if item.get("status") == "partial_success") overall_status = ( "success" if failed == 0 and pending_readback == 0 and partial == 0 else ("partial_success" if succeeded > 0 or pending_readback > 0 or any_write_executed or any_write_may_have_succeeded else "failed") ) uncertain_write = any_write_may_have_succeeded or pending_readback > 0 response: JSONObject = { "status": overall_status, "mode": "multi_app", "total": len(apps), "succeeded": succeeded, "failed": failed, "pending_readback": pending_readback, "created_app_keys": created_app_keys, "write_executed": any_write_executed, "safe_to_retry": not (any_write_executed or uncertain_write), "package_id": package_id, "publish_requested": publish, "apps": final_items, "parallelism": parallelism, "normalized_args": normalized_args, "verification": { "all_apps_succeeded": failed == 0, "created_app_count": len(created_app_keys), "pending_readback_count": pending_readback, }, "request_id": None, "error_code": None if overall_status != "failed" else "MULTI_APP_SCHEMA_APPLY_FAILED", "recoverable": overall_status != "success", "message": ( "multi-app schema apply needs readback before retry" if pending_readback > 0 else ("multi-app schema apply completed" if overall_status != "failed" else "multi-app schema apply failed") ), } if uncertain_write: response["write_may_have_succeeded"] = True response["next_action"] = "readback_before_retry" response["pending_readback_app_keys"] = list(dict.fromkeys(pending_readback_app_keys)) response["pending_readback_app_names"] = list(dict.fromkeys(pending_readback_app_names)) return finish(response) def _stabilize_deferred_relation_metadata( self, *, profile: str, source_app_key: str, deferred_add_fields: list[JSONObject], delay_seconds: float, ) -> list[JSONObject]: edit_context_result = self._finish_deferred_relation_edit_context(profile=profile, app_key=source_app_key) time.sleep(delay_seconds) app_keys = [source_app_key] for key in _multi_app_relation_target_app_keys(deferred_add_fields): if key and key not in app_keys: app_keys.append(key) readback: list[JSONObject] = [] for app_key in app_keys: try: result = self._facade.app_read_fields(profile=profile, app_key=app_key) readback.append( { "app_key": app_key, "status": result.get("status"), "field_count": len(result.get("fields") or []), **({"edit_context": edit_context_result} if app_key == source_app_key else {}), } ) except (QingflowApiError, RuntimeError) as error: api_error = _coerce_api_error(error) readback.append( { "app_key": app_key, "status": "failed", "transport_error": { "http_status": api_error.http_status, "backend_code": api_error.backend_code, "category": api_error.category, "request_id": api_error.request_id, }, } ) return readback def _finish_deferred_relation_edit_context(self, *, profile: str, app_key: str) -> JSONObject: try: version_result = self._facade.apps.app_get_edit_version_no(profile=profile, app_key=app_key).get("result") or {} raw_version = version_result.get("editVersionNo") or version_result.get("versionNo") or 1 edit_version_no = int(raw_version) self._facade.apps.app_edit_finished(profile=profile, app_key=app_key, payload={"editVersionNo": edit_version_no}) return {"status": "success", "edit_version_no": edit_version_no} except (QingflowApiError, RuntimeError, TypeError, ValueError) as error: api_error = _coerce_api_error(error if isinstance(error, (QingflowApiError, RuntimeError)) else RuntimeError(str(error))) return { "status": "failed", "message": api_error.message, "transport_error": { "http_status": api_error.http_status, "backend_code": api_error.backend_code, "category": api_error.category, "request_id": api_error.request_id, }, } def _app_schema_apply_once( self, *, profile: str, app_key: str = "", package_id: int | None = None, app_name: str = "", app_title: str = "", icon: str = "", color: str = "", visibility: JSONObject | None = None, publish: bool = True, add_fields: list[JSONObject], update_fields: list[JSONObject], remove_fields: list[JSONObject], inline_form_layout_sections: list[JSONObject] | None = None, allow_inline_layout_fallback: bool = True, ) -> JSONObject: """执行内部辅助逻辑。""" started_at = time.perf_counter() effective_app_name = app_name or app_title creating = not bool(str(app_key or "").strip()) and package_id is not None and bool(str(effective_app_name or "").strip()) system_field_failure = _reserved_system_field_name_failure( tool_name="app_schema_apply", profile=profile, app_key=app_key, package_id=package_id, app_name=effective_app_name, icon=icon, color=color, visibility=visibility, publish=publish, add_fields=add_fields, update_fields=update_fields, remove_fields=remove_fields, ) if system_field_failure is not None: return system_field_failure icon_failure = _validate_workspace_icon_for_builder( tool_name="app_schema_apply", icon=icon, color=color, creating=creating, ) if icon_failure is not None: return icon_failure try: parsed_add = [FieldPatch.model_validate(item) for item in add_fields or []] parsed_update = [FieldUpdatePatch.model_validate(item) for item in update_fields or []] parsed_remove = [FieldRemovePatch.model_validate(item) for item in remove_fields or []] parsed_inline_layout = [ LayoutSectionPatch.model_validate(item) for item in (inline_form_layout_sections or []) ] parsed_visibility = VisibilityPatch.model_validate(visibility) if visibility is not None else None except ValidationError as exc: return _validation_failure( str(exc), tool_name="app_schema_apply", exc=exc, suggested_next_call={ "tool_name": "app_schema_apply", "arguments": { "profile": profile, "app_key": str(app_key or ""), "package_id": package_id, "app_name": effective_app_name, "icon": str(icon or ""), "color": str(color or ""), "visibility": visibility, "publish": publish, "add_fields": add_fields or [{"name": "字段名称", "type": "text", "required": False}], "update_fields": update_fields or [], "remove_fields": remove_fields or [], }, }, ) normalized_args = { "app_key": str(app_key or ""), "package_id": package_id, "app_name": effective_app_name, "icon": str(icon or ""), "color": str(color or ""), "visibility": parsed_visibility.model_dump(mode="json") if parsed_visibility is not None else None, "publish": publish, "add_fields": [patch.model_dump(mode="json") for patch in parsed_add], "update_fields": [patch.model_dump(mode="json") for patch in parsed_update], "remove_fields": [patch.model_dump(mode="json") for patch in parsed_remove], **( { "inline_form_layout_sections": [ section.model_dump(mode="json", exclude_none=True) for section in parsed_inline_layout ] } if parsed_inline_layout else {} ), } result = _safe_tool_call( lambda: self._facade.app_schema_apply( profile=profile, app_key=str(app_key or ""), package_tag_id=package_id, app_name=effective_app_name, icon=str(icon or ""), color=str(color or ""), visibility=parsed_visibility, publish=publish, add_fields=parsed_add, update_fields=parsed_update, remove_fields=parsed_remove, inline_form_layout_sections=parsed_inline_layout, ), error_code="SCHEMA_APPLY_FAILED", normalized_args=normalized_args, suggested_next_call={"tool_name": "app_schema_apply", "arguments": {"profile": profile, **normalized_args}}, ) public_result = _publicize_package_fields(result) if inline_form_layout_sections and allow_inline_layout_fallback and not _inline_form_layout_was_handled(public_result): layout_started_at = time.perf_counter() public_result = self._apply_inline_form_layout_after_schema( profile=profile, schema_result=public_result, fallback_app_key=str(app_key or ""), publish=publish, sections=inline_form_layout_sections, ) _merge_duration_breakdown( public_result, inline_layout_ms=_elapsed_ms(layout_started_at), ) _merge_duration_breakdown(public_result, total_ms=_elapsed_ms(started_at)) return public_result def _apply_inline_form_layout_after_schema( self, *, profile: str, schema_result: JSONObject, fallback_app_key: str = "", publish: bool, sections: list[JSONObject], ) -> JSONObject: result = deepcopy(schema_result) if not sections: return result app_key = str(result.get("app_key") or result.get("appKey") or fallback_app_key or "").strip() result["inline_form_layout"] = { "requested": True, "section_count": len(sections), "applied": False, } if result.get("status") not in {"success", "partial_success"}: result["inline_form_layout"]["skipped_reason"] = "schema_apply_failed" verification = result.get("verification") if isinstance(result.get("verification"), dict) else {} verification = deepcopy(verification) verification["inline_form_layout_verified"] = False result["verification"] = verification return result if not app_key: result["status"] = "partial_success" result["error_code"] = result.get("error_code") or "INLINE_FORM_LAYOUT_APP_KEY_MISSING" result["message"] = "schema applied but inline form layout could not run because app_key was not resolved" result["inline_form_layout"]["error_code"] = "INLINE_FORM_LAYOUT_APP_KEY_MISSING" verification = result.get("verification") if isinstance(result.get("verification"), dict) else {} verification = deepcopy(verification) verification["inline_form_layout_verified"] = False result["verification"] = verification result["safe_to_retry"] = False return result layout_result = self._app_layout_apply_once( profile=profile, app_key=app_key, mode="merge", publish=publish, sections=sections, ) layout_ok = isinstance(layout_result, dict) and layout_result.get("status") in {"success", "partial_success"} result["inline_form_layout"] = { "requested": True, "section_count": len(sections), "app_key": app_key, "applied": layout_ok, "layout_status": layout_result.get("status") if isinstance(layout_result, dict) else "failed", **({"error_code": layout_result.get("error_code")} if isinstance(layout_result, dict) and layout_result.get("error_code") else {}), **({"message": layout_result.get("message")} if isinstance(layout_result, dict) and layout_result.get("message") else {}), } result["inline_form_layout_result"] = layout_result verification = result.get("verification") if isinstance(result.get("verification"), dict) else {} verification = deepcopy(verification) verification["inline_form_layout_verified"] = layout_ok result["verification"] = verification if layout_ok: result["write_executed"] = True return result result["status"] = "partial_success" if result.get("status") == "success" else result.get("status", "partial_success") result["error_code"] = result.get("error_code") or "INLINE_FORM_LAYOUT_APPLY_FAILED" result["message"] = "schema applied but inline form layout failed" result["safe_to_retry"] = False return result def app_layout_apply( self, *, profile: str, app_key: str, mode: str = "merge", publish: bool = True, sections: list[JSONObject], apps: list[JSONObject] | None = None, ) -> JSONObject: """执行应用相关逻辑。""" if apps is not None: return self._apply_app_batch( tool_name="app_layout_apply", profile=profile, apps=apps, apply_one=lambda _index, item, item_app_key: self.app_layout_apply( profile=profile, app_key=item_app_key, mode=str(_payload_get(item, "mode", default=mode) or mode), publish=_payload_bool(item, "publish", default=publish), sections=_payload_list(item, "sections", default=sections or []), apps=None, ), ) result = self._app_layout_apply_once( profile=profile, app_key=app_key, mode=mode, publish=publish, sections=sections, ) result = self._retry_after_self_lock_release( profile=profile, result=result, retry_call=lambda: self._app_layout_apply_once( profile=profile, app_key=app_key, mode=mode, publish=publish, sections=sections, ), ) return _attach_builder_apply_envelope("app_layout_apply", result) def _app_layout_apply_once(self, *, profile: str, app_key: str, mode: str = "merge", publish: bool = True, sections: list[JSONObject]) -> JSONObject: """执行内部辅助逻辑。""" try: parsed_mode = LayoutApplyMode(str(mode or "merge")) parsed_sections = [LayoutSectionPatch.model_validate(item) for item in sections or []] except (ValueError, ValidationError) as exc: return _validation_failure( str(exc), tool_name="app_layout_apply", exc=exc if isinstance(exc, ValidationError) else None, suggested_next_call={ "tool_name": "app_layout_apply", "arguments": { "profile": profile, "app_key": app_key, "mode": str(mode or "merge"), "publish": publish, "sections": sections or [{"title": "基础信息", "rows": [["字段A", "字段B"]]}], }, }, ) normalized_args = { "app_key": app_key, "mode": parsed_mode.value, "publish": publish, "sections": [section.model_dump(mode="json", exclude_none=True) for section in parsed_sections], } return _safe_tool_call( lambda: self._facade.app_layout_apply( profile=profile, app_key=app_key, mode=parsed_mode, publish=publish, sections=parsed_sections, ), error_code="LAYOUT_APPLY_FAILED", normalized_args=normalized_args, suggested_next_call={"tool_name": "app_layout_apply", "arguments": {"profile": profile, **normalized_args}}, ) def app_flow_apply( self, *, profile: str, app_key: str, mode: str | None = None, publish: bool = True, nodes: list[JSONObject] | None = None, transitions: list[JSONObject] | None = None, spec: JSONObject | None = None, idempotency_key: str | None = None, schema_version: str | None = None, patch_nodes: list[JSONObject] | None = None, ) -> JSONObject: """执行应用相关逻辑。""" if patch_nodes is not None: normalized_args = { "app_key": app_key, "publish": publish, "patch_nodes": patch_nodes, "idempotency_key": idempotency_key, "schema_version": schema_version, } return _attach_builder_apply_envelope("app_flow_apply", _safe_tool_call( lambda: self._facade.flow_patch_nodes( profile=profile, app_key=app_key, patch_nodes=patch_nodes, publish=publish, idempotency_key=idempotency_key, schema_version=schema_version, ), error_code="FLOW_PATCH_FAILED", normalized_args=normalized_args, suggested_next_call={"tool_name": "app_flow_get", "arguments": {"profile": profile, "app_key": app_key}}, )) if spec is not None: normalized_args = { "app_key": app_key, "publish": publish, "spec": spec, "idempotency_key": idempotency_key, "schema_version": schema_version, } return _attach_builder_apply_envelope("app_flow_apply", _safe_tool_call( lambda: self._facade.flow_apply( profile=profile, app_key=app_key, spec=spec, publish=publish, idempotency_key=idempotency_key, schema_version=schema_version, ), error_code="FLOW_SPEC_APPLY_FAILED", normalized_args=normalized_args, suggested_next_call={"tool_name": "app_flow_get", "arguments": {"profile": profile, "app_key": app_key}}, )) if nodes is not None or transitions is not None: effective_nodes = nodes or [] effective_transitions = transitions or [] result = self._app_flow_apply_once( profile=profile, app_key=app_key, mode=mode or "replace", publish=publish, nodes=effective_nodes, transitions=effective_transitions, ) result = self._retry_after_self_lock_release( profile=profile, result=result, retry_call=lambda: self._app_flow_apply_once( profile=profile, app_key=app_key, mode=mode or "replace", publish=publish, nodes=effective_nodes, transitions=effective_transitions, ), ) return _attach_builder_apply_envelope("app_flow_apply", result) normalized_args = {"app_key": app_key, "publish": publish} return _attach_builder_apply_envelope( "app_flow_apply", { "status": "failed", "error_code": "FLOW_SPEC_OR_PATCH_REQUIRED", "recoverable": True, "message": "app_flow_apply requires spec for a complete WorkflowSpec write or patch_nodes for targeted edits.", "normalized_args": normalized_args, "missing_fields": ["spec"], "allowed_values": {}, "details": {}, "suggested_next_call": {"tool_name": "app_flow_get_schema", "arguments": {"profile": profile}}, "request_id": None, "backend_code": None, "http_status": None, "noop": False, "warnings": [], "verification": {}, "verified": False, }, ) def _app_flow_apply_once( self, *, profile: str, app_key: str, mode: str = "replace", publish: bool = True, nodes: list[JSONObject], transitions: list[JSONObject], ) -> JSONObject: """执行内部辅助逻辑。""" try: request = FlowPlanRequest.model_validate( { "app_key": app_key, "mode": mode, "nodes": nodes or [], "transitions": transitions or [], "preset": None, } ) except ValidationError as exc: return _validation_failure( str(exc), tool_name="app_flow_apply", exc=exc, suggested_next_call={ "tool_name": "app_flow_apply", "arguments": { "profile": profile, "app_key": app_key, "mode": str(mode or "replace"), "publish": publish, "nodes": nodes or [{"id": "start", "type": "start", "name": "发起"}], "transitions": transitions or [], }, }, ) normalized_args = { "app_key": request.app_key, "mode": request.mode, "publish": publish, "nodes": [node.model_dump(mode="json") for node in request.nodes], "transitions": [transition.model_dump(mode="json", by_alias=True) for transition in request.transitions], } return _safe_tool_call( lambda: self._facade.app_flow_apply( profile=profile, app_key=request.app_key, mode=request.mode, publish=publish, nodes=[node.model_dump(mode="json") for node in request.nodes], transitions=[transition.model_dump(mode="json", by_alias=True) for transition in request.transitions], ), error_code="FLOW_APPLY_FAILED", normalized_args=normalized_args, suggested_next_call={"tool_name": "app_flow_apply", "arguments": {"profile": profile, **normalized_args}}, ) def app_views_apply( self, *, profile: str, app_key: str, publish: bool = True, upsert_views: list[JSONObject], patch_views: list[JSONObject] | None = None, remove_views: list[str], apps: list[JSONObject] | None = None, views: list[JSONObject] | None = None, ) -> JSONObject: """执行应用相关逻辑。""" if views is not None: return self._app_views_apply_view_batch( profile=profile, publish=publish, views=views, ) if apps is not None: return self._apply_app_batch( tool_name="app_views_apply", profile=profile, apps=apps, apply_one=lambda _index, item, item_app_key: self.app_views_apply( profile=profile, app_key=item_app_key, publish=_payload_bool(item, "publish", default=publish), upsert_views=_payload_list(item, "upsert_views", "upsertViews", "views", default=upsert_views or []), patch_views=_payload_list(item, "patch_views", "patchViews", default=patch_views or []), remove_views=_payload_list(item, "remove_views", "removeViews", default=remove_views or []), apps=None, ), ) result = self._app_views_apply_once( profile=profile, app_key=app_key, publish=publish, upsert_views=upsert_views, patch_views=patch_views or [], remove_views=remove_views, ) result = self._retry_after_self_lock_release( profile=profile, result=result, retry_call=lambda: self._app_views_apply_once( profile=profile, app_key=app_key, publish=publish, remove_views=remove_views, upsert_views=upsert_views, patch_views=patch_views or [], ), ) return _attach_builder_apply_envelope("app_views_apply", result) def _app_views_apply_view_batch( self, *, profile: str, publish: bool, views: list[JSONObject], ) -> JSONObject: started_at = time.perf_counter() if not isinstance(views, list) or not views: return _attach_builder_apply_envelope( "app_views_apply", _config_failure( tool_name="app_views_apply", message="app_views_apply views[] batch mode requires a non-empty views array.", fix_hint="Pass views as a JSON array; each item must include app_key and operation.", details={"expected_shape": {"views": [{"operation": "upsert", "app_key": "APP_KEY", "name": "业务视图", "type": "table", "columns": ["字段A"]}]}}, ), ) def _infer_operation(item: JSONObject) -> str: raw = str(item.get("operation") or item.get("op") or "").strip().lower() aliases = { "create": "upsert", "update": "upsert", "replace": "upsert", "delete": "remove", "remove": "remove", "patch": "patch", "upsert": "upsert", } if raw: return aliases.get(raw, raw) if "set" in item or "unset" in item or "update" in item or "values" in item: return "patch" if "view_key" in item or "viewKey" in item: return "patch" if ("name" in item or "type" in item or "columns" in item or "fields" in item) else "remove" return "upsert" def _view_job_payload(item: JSONObject) -> JSONObject: return { key: deepcopy(value) for key, value in item.items() if key not in { "operation", "op", "app_key", "appKey", "publish", } } def _view_selector(payload: JSONObject) -> str: return str(payload.get("view_key") or payload.get("viewKey") or payload.get("viewgraphKey") or payload.get("viewGraphKey") or payload.get("name") or "").strip() normalized_jobs: list[JSONObject] = [] job_results: list[JSONObject | None] = [None for _ in views] errors: list[JSONObject] = [] seen_patch_remove: dict[tuple[str, str], int] = {} seen_upsert_names: dict[tuple[str, str], int] = {} def _add_validation_error(index: int, error_code: str, message: str, *, app_key: str = "", operation: str = "", reason_path: str = "") -> None: error = { "index": index, "app_key": app_key, "operation": operation, "status": "failed", "error_code": error_code, "message": message, **({"reason_path": reason_path} if reason_path else {}), } errors.append(error) job_results[index] = error for index, item in enumerate(views): if not isinstance(item, dict): _add_validation_error(index, "VIEWS_FILE_ITEM_INVALID", "views[] item must be an object", reason_path=f"views[{index}]") continue app_key = str(item.get("app_key") or item.get("appKey") or "").strip() operation = _infer_operation(item) payload = _view_job_payload(item) if not app_key: _add_validation_error(index, "VIEWS_FILE_APP_KEY_REQUIRED", "views[] item requires app_key", operation=operation, reason_path=f"views[{index}].app_key") continue if operation not in {"upsert", "patch", "remove"}: _add_validation_error(index, "VIEWS_FILE_OPERATION_INVALID", "views[].operation must be upsert, patch, or remove", app_key=app_key, operation=operation, reason_path=f"views[{index}].operation") continue selector = _view_selector(payload) if operation in {"patch", "remove"} and not selector: _add_validation_error(index, "VIEWS_FILE_VIEW_SELECTOR_REQUIRED", f"{operation} view job requires view_key or name", app_key=app_key, operation=operation, reason_path=f"views[{index}].view_key") continue if operation in {"patch", "remove"} and selector: conflict_key = (app_key, selector) first_index = seen_patch_remove.get(conflict_key) if first_index is not None: _add_validation_error(index, "DUPLICATE_VIEW_JOB_CONFLICT", f"views[{index}] conflicts with views[{first_index}] for the same app/view selector", app_key=app_key, operation=operation, reason_path=f"views[{index}]") continue seen_patch_remove[conflict_key] = index if operation == "upsert": name = str(payload.get("name") or "").strip() if name: conflict_key = (app_key, name) first_index = seen_upsert_names.get(conflict_key) if first_index is not None: _add_validation_error(index, "DUPLICATE_VIEW_NAME_IN_BATCH", f"views[{index}] duplicates upsert view name from views[{first_index}] in the same app", app_key=app_key, operation=operation, reason_path=f"views[{index}].name") continue seen_upsert_names[conflict_key] = index normalized_jobs.append( { "index": index, "app_key": app_key, "operation": operation, "publish": _payload_bool(item, "publish", default=publish), "payload": payload, } ) def _extract_view_identity(result: JSONObject, fallback_payload: JSONObject, fallback_operation: str) -> tuple[str | None, str | None, str]: diff = result.get("views_diff") if isinstance(result.get("views_diff"), dict) else {} for key, operation_name in (("created", "created"), ("updated", "updated"), ("removed", "removed"), ("failed", "failed")): entries = diff.get(key) if isinstance(entries, list) and entries: entry = entries[0] if isinstance(entry, dict): return ( str(entry.get("name") or entry.get("view_name") or entry.get("viewName") or fallback_payload.get("name") or "").strip() or None, str(entry.get("view_key") or entry.get("viewKey") or fallback_payload.get("view_key") or fallback_payload.get("viewKey") or "").strip() or None, operation_name, ) if entry not in (None, ""): return (str(entry), None, operation_name) fallback_name = str(fallback_payload.get("name") or "").strip() or None fallback_view_key = str(fallback_payload.get("view_key") or fallback_payload.get("viewKey") or "").strip() or None return fallback_name, fallback_view_key, "removed" if fallback_operation == "remove" else "updated" def _run_view_job(job: JSONObject) -> JSONObject: payload = deepcopy(job["payload"]) operation = str(job["operation"]) app_key = str(job["app_key"]) job_publish = bool(job.get("publish")) if operation == "upsert": call_kwargs = {"upsert_views": [payload], "patch_views": [], "remove_views": []} elif operation == "patch": call_kwargs = {"upsert_views": [], "patch_views": [payload], "remove_views": []} else: call_kwargs = {"upsert_views": [], "patch_views": [], "remove_views": [_view_selector(payload)]} result = self._app_views_apply_once( profile=profile, app_key=app_key, publish=job_publish, **call_kwargs, ) result = self._retry_after_self_lock_release( profile=profile, result=result, retry_call=lambda: self._app_views_apply_once( profile=profile, app_key=app_key, publish=job_publish, **call_kwargs, ), ) status = str(result.get("status") or "success") if isinstance(result, dict) else "failed" name, view_key, actual_operation = _extract_view_identity(result if isinstance(result, dict) else {}, payload, operation) item: JSONObject = { "index": job["index"], "app_key": app_key, "operation": actual_operation if status != "failed" else operation, "requested_operation": operation, "status": status, **({"name": name} if name else {}), **({"view_key": view_key} if view_key else {}), "result": result if isinstance(result, dict) else {"status": "failed", "message": str(result)}, } if status == "failed": item["error_code"] = item["result"].get("error_code") item["message"] = item["result"].get("message") return item view_write_started_at = time.perf_counter() view_workers = min(VIEW_APPLY_PARALLEL_LIMIT, len(normalized_jobs)) if normalized_jobs else 0 if normalized_jobs: with ThreadPoolExecutor(max_workers=view_workers) as executor: future_map = {executor.submit(_run_view_job, job): job for job in normalized_jobs} for future in as_completed(future_map): job = future_map[future] try: item_result = future.result() except (QingflowApiError, RuntimeError) as error: api_error = _coerce_api_error(error) item_result = { "index": job["index"], "app_key": job["app_key"], "operation": job["operation"], "requested_operation": job["operation"], "status": "failed", "error_code": "VIEW_BATCH_JOB_FAILED", "message": _public_error_message("VIEW_BATCH_JOB_FAILED", api_error), "details": {"transport_error": _transport_error_payload(api_error)}, "request_id": api_error.request_id, "backend_code": api_error.backend_code, "http_status": None if api_error.http_status == 404 else api_error.http_status, } job_results[int(job["index"])] = item_result view_write_parallel_wall_ms = _elapsed_ms(view_write_started_at) final_views = [item if isinstance(item, dict) else {"index": index, "status": "failed", "error_code": "VIEW_BATCH_JOB_MISSING", "message": "view batch job did not produce a result"} for index, item in enumerate(job_results)] errors.extend(item for item in final_views if str(item.get("status") or "") == "failed" and item not in errors) partial = sum(1 for item in final_views if str(item.get("status") or "") == "partial_success") succeeded = sum(1 for item in final_views if str(item.get("status") or "") in {"success", "partial_success"}) failed = len(final_views) - succeeded status = "failed" if succeeded == 0 else "partial_success" if failed > 0 or partial > 0 else "success" write_executed = any( isinstance(item.get("result"), dict) and bool(item["result"].get("write_executed")) for item in final_views ) payload: JSONObject = { "status": status, "error_code": None if status == "success" else "APP_VIEWS_APPLY_BATCH_PARTIAL" if succeeded else "APP_VIEWS_APPLY_BATCH_FAILED", "recoverable": failed > 0 or partial > 0, "message": f"applied app_views_apply to {succeeded}/{len(final_views)} views" if status != "success" else f"applied app_views_apply to {succeeded} views", "normalized_args": {"views": views}, "views": final_views, "errors": errors, "parallelism": {"view_workers": view_workers, "same_app_parallel": True}, "verification": { "batch_verified": failed == 0 and partial == 0, "succeeded": succeeded, "partial": partial, "failed": failed, }, "write_executed": write_executed, "write_succeeded": write_executed and failed == 0 and partial == 0, "safe_to_retry": not write_executed, } _merge_duration_breakdown( payload, view_write_parallel_wall_ms=view_write_parallel_wall_ms, total_ms=_elapsed_ms(started_at), ) return _attach_builder_apply_envelope("app_views_apply", payload) def _app_views_apply_once( self, *, profile: str, app_key: str, publish: bool = True, upsert_views: list[JSONObject], patch_views: list[JSONObject], remove_views: list[str], ) -> JSONObject: """执行内部辅助逻辑。""" reserved_failure = _reserved_system_view_name_failure( tool_name="app_views_apply", profile=profile, app_key=app_key, publish=publish, upsert_views=upsert_views, patch_views=patch_views, remove_views=remove_views, ) if reserved_failure is not None: return reserved_failure try: parsed_views = [ViewUpsertPatch.model_validate(item) for item in (upsert_views or [])] parsed_patch_views = [ViewPartialPatch.model_validate(item) for item in (patch_views or [])] except ValidationError as exc: return _visibility_validation_failure( str(exc), tool_name="app_views_apply", exc=exc, suggested_next_call={ "tool_name": "app_views_apply", "arguments": { "profile": profile, "app_key": app_key, "publish": publish, "upsert_views": upsert_views or [{"name": "业务台账视图", "type": "table", "columns": ["字段A"]}], "patch_views": patch_views or [], "remove_views": remove_views or [], }, }, ) normalized_args = { "app_key": app_key, "publish": publish, "upsert_views": [public_view_upsert_payload(view) for view in parsed_views], "patch_views": [public_view_partial_payload(patch) for patch in parsed_patch_views], "remove_views": list(remove_views or []), } return _safe_tool_call( lambda: self._facade.app_views_apply( profile=profile, app_key=app_key, publish=publish, upsert_views=parsed_views, patch_views=parsed_patch_views, remove_views=list(remove_views or []), ), error_code="VIEWS_APPLY_FAILED", normalized_args=normalized_args, suggested_next_call={"tool_name": "app_views_apply", "arguments": {"profile": profile, **normalized_args}}, ) def chart_apply( self, *, profile: str, app_key: str, upsert_charts: list[JSONObject], patch_charts: list[JSONObject] | None = None, remove_chart_ids: list[str], reorder_chart_ids: list[str], ) -> JSONObject: """执行图表相关逻辑。""" return self.app_charts_apply( profile=profile, app_key=app_key, upsert_charts=upsert_charts, patch_charts=patch_charts or [], remove_chart_ids=remove_chart_ids, reorder_chart_ids=reorder_chart_ids, ) def app_charts_apply( self, *, profile: str, app_key: str, upsert_charts: list[JSONObject], remove_chart_ids: list[str], reorder_chart_ids: list[str], patch_charts: list[JSONObject] | None = None, apps: list[JSONObject] | None = None, ) -> JSONObject: """执行应用相关逻辑。""" if apps is not None: return self._apply_app_batch( tool_name="app_charts_apply", profile=profile, apps=apps, apply_one=lambda _index, item, item_app_key: self.app_charts_apply( profile=profile, app_key=item_app_key, upsert_charts=_payload_list(item, "upsert_charts", "upsertCharts", "charts", default=upsert_charts or []), patch_charts=_payload_list(item, "patch_charts", "patchCharts", default=patch_charts or []), remove_chart_ids=_payload_list(item, "remove_chart_ids", "removeChartIds", default=remove_chart_ids or []), reorder_chart_ids=_payload_list(item, "reorder_chart_ids", "reorderChartIds", default=reorder_chart_ids or []), apps=None, ), ) try: request = ChartApplyRequest.model_validate( { "app_key": app_key, "upsert_charts": upsert_charts or [], "patch_charts": patch_charts or [], "remove_chart_ids": remove_chart_ids or [], "reorder_chart_ids": reorder_chart_ids or [], } ) except ValidationError as exc: return _attach_builder_apply_envelope("app_charts_apply", _visibility_validation_failure( str(exc), tool_name="app_charts_apply", exc=exc, suggested_next_call={ "tool_name": "app_charts_apply", "arguments": { "profile": profile, "app_key": app_key, "upsert_charts": [{"name": "销售总量", "chart_type": "target", "metric": "count(*)"}], "remove_chart_ids": [], "reorder_chart_ids": [], }, }, )) normalized_args = request.model_dump(mode="json") return _attach_builder_apply_envelope("app_charts_apply", _safe_tool_call( lambda: self._facade.chart_apply(profile=profile, request=request), error_code="CHART_APPLY_FAILED", normalized_args=normalized_args, suggested_next_call={"tool_name": "app_charts_apply", "arguments": {"profile": profile, **normalized_args}}, )) def portal_apply( self, *, profile: str, dash_key: str = "", dash_name: str = "", name: str = "", package_id: int | None = None, publish: bool = True, sections: list[JSONObject] | None = None, pages: list[JSONObject] | None = None, layout_preset: str = "", visibility: JSONObject | None = None, auth: JSONObject | None = None, icon: str | None = None, color: str | None = None, hide_copyright: bool | None = None, dash_global_config: JSONObject | None = None, config: JSONObject | None = None, payload: JSONObject | None = None, patch_sections: list[JSONObject] | None = None, ) -> JSONObject: """执行门户相关逻辑。""" if patch_sections is not None: normalized_args = { "dash_key": dash_key, "publish": publish, "patch_sections": patch_sections, } return _attach_builder_apply_envelope("portal_apply", _publicize_package_fields(_safe_tool_call( lambda: self._facade.portal_patch_sections( profile=profile, dash_key=dash_key, patch_sections=patch_sections, publish=publish, ), error_code="PORTAL_PATCH_FAILED", normalized_args=normalized_args, suggested_next_call={"tool_name": "portal_get", "arguments": {"profile": profile, "dash_key": dash_key}}, ))) request_payload: dict[str, Any] = dict(payload) if isinstance(payload, dict) else {} if dash_key: request_payload["dash_key"] = dash_key if dash_name: request_payload["dash_name"] = dash_name elif name: request_payload["name"] = name if package_id is not None: request_payload["package_id"] = package_id if "publish" not in request_payload or publish is False: request_payload["publish"] = publish if sections: request_payload["sections"] = sections if pages: request_payload["pages"] = pages if layout_preset: request_payload["layout_preset"] = layout_preset if visibility is not None: request_payload["visibility"] = visibility if auth is not None: request_payload["auth"] = auth if icon is not None: request_payload["icon"] = icon if color is not None: request_payload["color"] = color if hide_copyright is not None: request_payload["hide_copyright"] = hide_copyright if dash_global_config is not None: request_payload["dash_global_config"] = dash_global_config if config: merged_config = dict(request_payload.get("config") or {}) if isinstance(request_payload.get("config"), dict) else {} merged_config.update(config) request_payload["config"] = merged_config try: request = PortalApplyRequest.model_validate(request_payload) except ValidationError as exc: return _attach_builder_apply_envelope("portal_apply", _visibility_validation_failure( str(exc), tool_name="portal_apply", exc=exc, suggested_next_call={ "tool_name": "portal_apply", "arguments": { "profile": profile, "dash_name": dash_name or "业务门户", "package_id": package_id or 1001, "publish": True, "layout_preset": "dashboard_2col", "sections": [ { "title": "经营概览", "source_type": "text", "text": "欢迎使用业务门户", } ], }, }, )) normalized_args = request.model_dump(mode="json") normalized_args["package_id"] = normalized_args.pop("package_tag_id", package_id) icon_failure = _validate_workspace_icon_for_builder( tool_name="portal_apply", icon=str(request.icon or ""), color=str(request.color or ""), creating=not bool(str(request.dash_key or "").strip()), ) if icon_failure is not None: return _attach_builder_apply_envelope("portal_apply", icon_failure) result = _publicize_package_fields(_safe_tool_call( lambda: self._facade.portal_apply(profile=profile, request=request), error_code="PORTAL_APPLY_FAILED", normalized_args=normalized_args, suggested_next_call={"tool_name": "portal_apply", "arguments": {"profile": profile, **normalized_args}}, )) return _attach_builder_apply_envelope("portal_apply", result) def portal_delete(self, *, profile: str, dash_key: str) -> JSONObject: """执行门户删除逻辑。""" normalized_args = {"dash_key": dash_key} result = _publicize_package_fields(_safe_tool_call( lambda: self._facade.portal_delete(profile=profile, dash_key=dash_key), error_code="PORTAL_DELETE_FAILED", normalized_args=normalized_args, suggested_next_call={"tool_name": "portal_delete", "arguments": {"profile": profile, **normalized_args}}, )) return _attach_builder_apply_envelope("portal_delete", result) def app_publish_verify( self, *, profile: str, app_key: str, app_keys: list[str] | None = None, expected_package_id: int | None = None, ) -> JSONObject: """执行应用相关逻辑。""" if app_keys is not None: return self._apply_app_batch( tool_name="app_publish_verify", profile=profile, apps=[{"app_key": item, "expected_package_id": expected_package_id} for item in app_keys], apply_one=lambda _index, item, item_app_key: self.app_publish_verify( profile=profile, app_key=item_app_key, app_keys=None, expected_package_id=item.get("expected_package_id") or item.get("expectedPackageId") or expected_package_id, ), ) normalized_args = {"app_key": app_key, "expected_package_id": expected_package_id} result = _publicize_package_fields(_safe_tool_call( lambda: self._facade.app_publish_verify(profile=profile, app_key=app_key, expected_package_tag_id=expected_package_id), error_code="PUBLISH_VERIFY_FAILED", normalized_args=normalized_args, suggested_next_call={"tool_name": "app_publish_verify", "arguments": {"profile": profile, **normalized_args}}, )) result = _publicize_package_fields(self._retry_after_self_lock_release( profile=profile, result=result, retry_call=lambda: self._facade.app_publish_verify( profile=profile, app_key=app_key, expected_package_tag_id=expected_package_id, ), )) return _attach_builder_apply_envelope("app_publish_verify", result) def _retry_after_self_lock_release(self, *, profile: str, result: JSONObject, retry_call) -> JSONObject: """执行内部辅助逻辑。""" if not isinstance(result, dict) or result.get("status") != "failed" or result.get("error_code") != "APP_EDIT_LOCKED": return result suggested = result.get("suggested_next_call") if not isinstance(suggested, dict) or suggested.get("tool_name") != "app_release_edit_lock_if_mine": return result arguments = suggested.get("arguments") if not isinstance(arguments, dict): return result app_key = str(arguments.get("app_key") or "") lock_owner_email = str(arguments.get("lock_owner_email") or "") lock_owner_name = str(arguments.get("lock_owner_name") or "") release_attempts: list[JSONObject] = [] retried: JSONObject = result for _ in range(3): release_result = self.app_release_edit_lock_if_mine( profile=profile, app_key=app_key, lock_owner_email=lock_owner_email, lock_owner_name=lock_owner_name, ) release_attempts.append(release_result) if not isinstance(release_result, dict) or release_result.get("status") != "success": result.setdefault("details", {}) if isinstance(result["details"], dict): result["details"]["edit_lock_release_result"] = release_result result["details"]["edit_lock_release_attempts"] = release_attempts return result retried = retry_call() if not ( isinstance(retried, dict) and retried.get("status") == "failed" and retried.get("error_code") == "APP_EDIT_LOCKED" ): break time.sleep(0.2) if ( isinstance(retried, dict) and retried.get("status") == "failed" and retried.get("error_code") == "APP_EDIT_LOCKED" ): retried = { **retried, "error_code": "PERSISTENT_SELF_LOCK", "message": "app remains locked by the current user's active editor session after repeated forced release attempts", "recoverable": True, "suggested_next_call": None, } if isinstance(retried, dict): retried.setdefault("details", {}) if isinstance(retried["details"], dict): retried["details"]["edit_lock_release_result"] = release_attempts[-1] if release_attempts else None retried["details"]["edit_lock_release_attempts"] = release_attempts retried["edit_lock_released"] = bool(release_attempts) retried["retried_after_edit_lock_release"] = True return retried def _rewrite_plan_result_for_apply( self, *, result: JSONObject, profile: str, publish: bool, plan_tool_name: str, apply_tool_name: str, ) -> JSONObject: """执行内部辅助逻辑。""" if not isinstance(result, dict): return result rewritten = dict(result) if rewritten.get("error_code") == "VALIDATION_ERROR": contract = _BUILDER_TOOL_CONTRACTS.get(apply_tool_name) details = rewritten.get("details") if not isinstance(details, dict): details = {} rewritten["details"] = details if isinstance(contract, dict): rewritten["allowed_values"] = deepcopy(contract.get("allowed_values", {})) details["allowed_keys"] = deepcopy(contract.get("allowed_keys", [])) details["section_allowed_keys"] = deepcopy(contract.get("section_allowed_keys", [])) details["section_aliases"] = deepcopy(contract.get("section_aliases", {})) details["minimal_section_example"] = deepcopy(contract.get("minimal_section_example")) suggested_next_call = rewritten.get("suggested_next_call") if isinstance(suggested_next_call, dict): if suggested_next_call.get("tool_name") == plan_tool_name: arguments = suggested_next_call.get("arguments") normalized_arguments = dict(arguments) if isinstance(arguments, dict) else {} normalized_arguments.setdefault("profile", profile) normalized_arguments["publish"] = publish rewritten["suggested_next_call"] = { **suggested_next_call, "tool_name": apply_tool_name, "arguments": normalized_arguments, } return rewritten if rewritten.get("error_code") == "VALIDATION_ERROR" and suggested_next_call.get("tool_name") == apply_tool_name: arguments = suggested_next_call.get("arguments") normalized_arguments = dict(arguments) if isinstance(arguments, dict) else {} normalized_arguments.setdefault("profile", profile) normalized_arguments["publish"] = publish if isinstance(details, dict): details["canonical_arguments"] = normalized_arguments rewritten["suggested_next_call"] = { **suggested_next_call, "arguments": normalized_arguments, } if rewritten.get("status") == "success": normalized_args = rewritten.get("normalized_args") if isinstance(normalized_args, dict): rewritten["suggested_next_call"] = { "tool_name": apply_tool_name, "arguments": {"profile": profile, **normalized_args, "publish": publish}, } return rewritten def _multi_app_item_failure(index: int, item: object, error_code: str, message: str) -> JSONObject: app_name = None client_key = None app_key = None if isinstance(item, dict): app_name = str(item.get("app_name") or item.get("appTitle") or item.get("app_title") or item.get("title") or "").strip() or None client_key = str(item.get("client_key") or item.get("clientKey") or "").strip() or None app_key = str(item.get("app_key") or item.get("appKey") or "").strip() or None return { "index": index, "row_number": index + 1, "client_key": client_key, "app_name": app_name, "app_key": app_key, "status": "failed", "stage": "validate_item", "error_code": error_code, "message": message, "safe_to_retry": True, } def _normalize_builder_icon_args( *, icon: str | JSONObject | None, color: str | None, icon_name: str | None = None, icon_color: str | None = None, icon_config: JSONObject | None = None, ) -> tuple[str | None, str | None]: payloads: list[JSONObject] = [] if isinstance(icon_config, dict): payloads.append(icon_config) if isinstance(icon, dict): payloads.append(icon) normalized_icon = None if isinstance(icon, dict) else icon normalized_color = color for payload in payloads: normalized_icon = normalized_icon or payload.get("icon") or payload.get("icon_name") or payload.get("iconName") or payload.get("name") normalized_color = normalized_color or payload.get("color") or payload.get("icon_color") or payload.get("iconColor") normalized_icon = normalized_icon or icon_name normalized_color = normalized_color or icon_color return ( str(normalized_icon).strip() if normalized_icon is not None else None, str(normalized_color).strip() if normalized_color is not None else None, ) def _normalize_schema_app_item(item: JSONObject) -> JSONObject: normalized = deepcopy(item) icon, color = _normalize_builder_icon_args( icon=normalized.get("icon"), color=normalized.get("color"), icon_name=normalized.get("icon_name") or normalized.get("iconName"), icon_color=normalized.get("icon_color") or normalized.get("iconColor"), icon_config=normalized.get("icon_config") or normalized.get("iconConfig"), ) if icon: normalized["icon"] = icon if color: normalized["color"] = color for key in ("icon_name", "iconName", "icon_color", "iconColor", "icon_config", "iconConfig"): normalized.pop(key, None) return normalized def _schema_apps_expected_shape() -> JSONObject: return { "package_id": 1001, "apps": [ { "client_key": "employee", "app_name": "员工花名册", "icon": "business-personalcard", "color": "emerald", "form": [ { "section": "基础信息", "rows": [[{"name": "员工名称", "type": "text", "data_title": True}]], } ], } ], } def _schema_apps_expected_shape_json() -> str: return ( '{"package_id":1001,"apps":[{"client_key":"employee","app_name":"员工花名册",' '"icon":"business-personalcard","color":"emerald","form":[{"section":"基础信息","rows":[[{"name":"员工名称","type":"text","data_title":true}]]}]}]}' ) def _is_schema_apps_wrapper_item(item: object) -> bool: return isinstance(item, dict) and "apps" in item def _schema_apps_shape_failure(*, tool_name: str, apps: list[JSONObject]) -> JSONObject | None: for index, item in enumerate(apps): if _is_schema_apps_wrapper_item(item): return _config_failure( tool_name=tool_name, error_code="APPS_FILE_SHAPE_INVALID", message="apps[] items must be app schema items, not package wrapper objects.", fix_hint=( "For MCP pass package_id separately and apps=[{app_name, icon, color, add_fields}]. " 'For CLI use --apps-file with {"package_id":1001,"apps":[...]}. ' "Do not pass multiple {package_id, apps} wrapper objects inside apps[]." ), details={ "index": index, "row_number": index + 1, "expected_shape": _schema_apps_expected_shape(), "expected_shape_json": _schema_apps_expected_shape_json(), }, ) return None def _normalize_schema_apps_argument(*, tool_name: str, package_id: int | None, apps: list[JSONObject]) -> JSONObject: normalized_package_id = package_id normalized_apps: list[JSONObject] = apps warnings: list[JSONObject] = [] if len(apps) == 1 and _is_schema_apps_wrapper_item(apps[0]): wrapper = apps[0] wrapper_apps = wrapper.get("apps") if not isinstance(wrapper_apps, list): return { "failure": _config_failure( tool_name=tool_name, error_code="APPS_FILE_SHAPE_INVALID", message="apps singleton wrapper requires an apps array.", fix_hint='Use {"package_id":1001,"apps":[...]} in CLI, or pass MCP package_id=1001 and apps=[...].', details={ "expected_shape": _schema_apps_expected_shape(), "expected_shape_json": _schema_apps_expected_shape_json(), }, ) } if normalized_package_id is None and wrapper.get("package_id") is not None: try: normalized_package_id = int(wrapper.get("package_id")) except (TypeError, ValueError): return { "failure": _config_failure( tool_name=tool_name, error_code="APPS_FILE_SHAPE_INVALID", message="apps singleton wrapper package_id must be an integer.", fix_hint="Pass package_id as a number at the top level.", details={ "expected_shape": _schema_apps_expected_shape(), "expected_shape_json": _schema_apps_expected_shape_json(), }, ) } normalized_apps = wrapper_apps warnings.append( { "code": "APPS_FILE_WRAPPER_ARRAY_UNWRAPPED", "message": "apps was a singleton wrapper array; normalized it to package_id + apps.", } ) else: failure = _schema_apps_shape_failure(tool_name=tool_name, apps=apps) if failure is not None: return {"failure": failure} normalized_items: list[JSONObject] = [] for item in normalized_apps: normalized_items.append(_normalize_schema_app_item(item) if isinstance(item, dict) else item) return {"package_id": normalized_package_id, "apps": normalized_items, "warnings": warnings} def _normalize_schema_form_field(field: JSONObject) -> JSONObject: payload = deepcopy(field) if "data_title" in payload and "as_data_title" not in payload and "asDataTitle" not in payload: payload["as_data_title"] = payload.pop("data_title") if "dataTitle" in payload and "as_data_title" not in payload and "asDataTitle" not in payload: payload["as_data_title"] = payload.pop("dataTitle") if "data_cover" in payload and "as_data_cover" not in payload and "asDataCover" not in payload: payload["as_data_cover"] = payload.pop("data_cover") if "dataCover" in payload and "as_data_cover" not in payload and "asDataCover" not in payload: payload["as_data_cover"] = payload.pop("dataCover") return payload def _schema_form_entry_name(entry: object) -> str: if isinstance(entry, dict): for key in ("name", "title", "label", "field_name", "fieldName"): if entry.get(key) is not None: return str(entry.get(key) or "").strip() return "" if isinstance(entry, (str, int)): return str(entry).strip() return "" def _schema_form_failure(*, tool_name: str, path: str, message: str, fix_hint: str) -> JSONObject: return _config_failure( tool_name=tool_name, error_code="FORM_SHAPE_INVALID", message=message, fix_hint=fix_hint, details={ "path": path, "expected_shape": { "form": [ { "section": "基础信息", "rows": [ [ {"name": "标题", "type": "text", "data_title": True}, {"name": "状态", "type": "single_select", "options": ["待处理", "处理中", "已完成"]}, ] ], } ] }, }, ) def _compile_schema_form_payload(*, tool_name: str, form: object, path: str = "form") -> JSONObject: if form is None: return {"add_fields": [], "layout_sections": []} if not isinstance(form, list) or not form: return { "failure": _schema_form_failure( tool_name=tool_name, path=path, message=f"{path} must be a non-empty list of form sections.", fix_hint='Use form=[{"section":"基础信息","rows":[[{"name":"标题","type":"text","data_title":true}]]}].', ) } add_fields: list[JSONObject] = [] layout_sections: list[JSONObject] = [] for section_index, section in enumerate(form): section_path = f"{path}[{section_index}]" if not isinstance(section, dict): return { "failure": _schema_form_failure( tool_name=tool_name, path=section_path, message=f"{section_path} must be an object.", fix_hint="Each form section must include section and rows.", ) } section_title = str(section.get("section") or section.get("title") or section.get("name") or "").strip() rows = section.get("rows") if not section_title: return { "failure": _schema_form_failure( tool_name=tool_name, path=f"{section_path}.section", message=f"{section_path}.section is required.", fix_hint='Set a business section title such as "基础信息" or "生产信息".', ) } if not isinstance(rows, list) or not rows: return { "failure": _schema_form_failure( tool_name=tool_name, path=f"{section_path}.rows", message=f"{section_path}.rows must be a non-empty list.", fix_hint="Use rows as a list of field rows; each row is a list of field objects.", ) } layout_rows: list[list[str]] = [] for row_index, row in enumerate(rows): row_path = f"{section_path}.rows[{row_index}]" if isinstance(row, dict): row_fields = row.get("fields") else: row_fields = row if not isinstance(row_fields, list) or not row_fields: return { "failure": _schema_form_failure( tool_name=tool_name, path=row_path, message=f"{row_path} must be a non-empty field list.", fix_hint='Use rows like [[{"name":"字段A","type":"text"},{"name":"字段B","type":"number"}]].', ) } layout_row: list[str] = [] for field_index, field in enumerate(row_fields): field_path = f"{row_path}[{field_index}]" field_name = _schema_form_entry_name(field) if not field_name: return { "failure": _schema_form_failure( tool_name=tool_name, path=field_path, message=f"{field_path} must include a field name.", fix_hint='Each new field object must include name, for example {"name":"工单编号","type":"text"}.', ) } layout_row.append(field_name) if isinstance(field, dict): add_fields.append(_normalize_schema_form_field(field)) layout_rows.append(layout_row) layout_sections.append({"title": section_title, "rows": layout_rows}) return {"add_fields": add_fields, "layout_sections": layout_sections} def _apply_schema_form_to_payload(*, tool_name: str, payload: JSONObject, path: str) -> JSONObject: form = payload.get("form") if form is None: return {"payload": payload, "layout_sections": []} existing_add_fields = payload.get("add_fields") or payload.get("addFields") if isinstance(existing_add_fields, list) and existing_add_fields: return { "failure": _config_failure( tool_name=tool_name, error_code="FORM_WITH_ADD_FIELDS_CONFLICT", message=f"{path} cannot use form and add_fields together.", fix_hint="Use form as the single schema creation shape: put fields under form[].rows[][].", details={"path": path, "conflicting_keys": ["form", "add_fields"]}, ) } compiled = _compile_schema_form_payload(tool_name=tool_name, form=form, path=f"{path}.form") if isinstance(compiled.get("failure"), dict): return {"failure": compiled["failure"]} normalized = deepcopy(payload) normalized.pop("form", None) normalized["add_fields"] = compiled.get("add_fields") or [] normalized["_inline_form_layout_sections"] = compiled.get("layout_sections") or [] return {"payload": normalized, "layout_sections": compiled.get("layout_sections") or []} def _multi_app_item_app_name(item: JSONObject) -> str: return str(item.get("app_name") or item.get("appName") or item.get("appTitle") or item.get("app_title") or item.get("title") or "").strip() def _multi_app_item_app_key(item: JSONObject) -> str: return str(item.get("app_key") or item.get("appKey") or "").strip() def _multi_app_item_client_key(item: JSONObject) -> str: return str(item.get("client_key") or item.get("clientKey") or "").strip() def _field_name_for_static_validation(field: JSONObject) -> str: return str(field.get("name") or field.get("title") or field.get("label") or "").strip() def _field_type_for_static_validation(field: JSONObject) -> str: return str(field.get("type") or field.get("type_id") or field.get("typeId") or "").strip() def _is_data_title_field(field: JSONObject) -> bool: return bool(field.get("as_data_title") or field.get("asDataTitle") or field.get("data_title") or field.get("dataTitle")) def _collect_multi_app_target_refs(value: object, *, path: str) -> list[JSONObject]: refs: list[JSONObject] = [] if isinstance(value, list): for index, entry in enumerate(value): refs.extend(_collect_multi_app_target_refs(entry, path=f"{path}[{index}]")) return refs if not isinstance(value, dict): return refs for key, entry in value.items(): current_path = f"{path}.{key}" if key in {"target_app_ref", "targetAppRef", "target_app_client_key", "targetAppClientKey"}: refs.append({"kind": "target_app_ref", "value": str(entry or "").strip(), "path": current_path}) continue if key in {"target_app", "targetApp"}: refs.append({"kind": "target_app", "value": str(entry or "").strip(), "path": current_path}) continue refs.extend(_collect_multi_app_target_refs(entry, path=current_path)) return refs def _multi_app_static_validation_failure( *, tool_name: str, apps: list[JSONObject], ) -> JSONObject | None: issues: list[JSONObject] = [] client_key_indexes: dict[str, int] = {} app_name_indexes: dict[str, list[int]] = {} for index, item in enumerate(apps): if not isinstance(item, dict): continue app_name = _multi_app_item_app_name(item) client_key = _multi_app_item_client_key(item) if client_key: first_index = client_key_indexes.get(client_key) if first_index is not None: issues.append( { "index": index, "row_number": index + 1, "path": f"apps[{index}].client_key", "error_code": "DUPLICATE_CLIENT_KEY", "message": f"duplicate client_key '{client_key}' also appears at apps[{first_index}]", "fix_hint": "Use a unique stable client_key for each app item, then reference relations through target_app_ref.", "details": {"client_key": client_key, "first_index": first_index, "duplicate_index": index}, } ) else: client_key_indexes[client_key] = index if app_name: app_name_indexes.setdefault(app_name, []).append(index) for app_name, indexes in sorted(app_name_indexes.items()): if len(indexes) <= 1: continue issues.append( { "index": indexes[1], "row_number": indexes[1] + 1, "path": f"apps[{indexes[1]}].app_name", "error_code": "DUPLICATE_APP_NAME", "message": f"duplicate app_name '{app_name}' appears at apps{indexes}", "fix_hint": "Use unique app_name values in one multi-app payload, or use app_key for existing apps.", "details": {"app_name": app_name, "indexes": indexes}, } ) for index, item in enumerate(apps): if not isinstance(item, dict): continue app_key = _multi_app_item_app_key(item) app_name = _multi_app_item_app_name(item) new_app_item = not bool(app_key) if not app_key and not app_name: issues.append( { "index": index, "row_number": index + 1, "path": f"apps[{index}]", "error_code": "APP_SELECTOR_REQUIRED", "message": "apps[] item requires app_key or app_name", "fix_hint": "For new apps pass app_name; for existing apps pass app_key.", } ) continue add_fields = _multi_app_list_value(item, "add_fields", "addFields") if new_app_item: title_fields: list[tuple[int, JSONObject]] = [ (field_index, field) for field_index, field in enumerate(add_fields) if isinstance(field, dict) and _is_data_title_field(field) ] if not title_fields: issues.append( { "index": index, "row_number": index + 1, "path": f"apps[{index}].add_fields", "error_code": "MISSING_DATA_TITLE_FIELD", "message": "new apps must define exactly one top-level data title field", "fix_hint": "Mark one readable top-level field with as_data_title=true, for example {'name':'客户名称','type':'text','as_data_title':true}.", "details": {"suggested_field_names": [_field_name_for_static_validation(field) for field in add_fields[:8] if isinstance(field, dict)]}, } ) elif len(title_fields) > 1: issues.append( { "index": index, "row_number": index + 1, "path": f"apps[{index}].add_fields", "error_code": "MULTIPLE_DATA_TITLE_FIELDS", "message": "new apps can mark only one top-level field as data title", "fix_hint": "Keep as_data_title=true on exactly one top-level field.", "details": { "fields": [ { "field_index": field_index, "name": _field_name_for_static_validation(field), "type": _field_type_for_static_validation(field), } for field_index, field in title_fields ] }, } ) else: field_index, title_field = title_fields[0] if _field_type_for_static_validation(title_field) in {"subtable", "table"}: issues.append( { "index": index, "row_number": index + 1, "path": f"apps[{index}].add_fields[{field_index}]", "error_code": "INVALID_DATA_TITLE_FIELD", "message": "data title must be a top-level non-subtable field", "fix_hint": "Move as_data_title=true to a normal top-level text/number/date-like field.", "details": { "field_name": _field_name_for_static_validation(title_field), "field_type": _field_type_for_static_validation(title_field), }, } ) if _contains_multi_app_target_ref(title_field): issues.append( { "index": index, "row_number": index + 1, "path": f"apps[{index}].add_fields[{field_index}]", "error_code": "DATA_TITLE_FIELD_DEFERRED_BY_TARGET_REF", "message": "data title cannot be a relation field that depends on target_app_ref/target_app in the same multi-app call", "fix_hint": "Use a normal title field such as 客户名称/项目名称 as data title; keep relation fields as non-title fields.", "details": {"field_name": _field_name_for_static_validation(title_field)}, } ) for ref in _collect_multi_app_target_refs(item, path=f"apps[{index}]"): ref_value = str(ref.get("value") or "").strip() if not ref_value: issues.append( { "index": index, "row_number": index + 1, "path": ref.get("path"), "error_code": "TARGET_APP_REFERENCE_EMPTY", "message": "target app reference cannot be empty", "fix_hint": "Use target_app_ref with another apps[].client_key, or target_app with a unique apps[].app_name.", } ) elif ref.get("kind") == "target_app_ref" and ref_value not in client_key_indexes: issues.append( { "index": index, "row_number": index + 1, "path": ref.get("path"), "error_code": "TARGET_APP_REF_UNRESOLVED", "message": f"target_app_ref '{ref_value}' does not match any apps[].client_key in this payload", "fix_hint": "Set target_app_ref to one of apps[].client_key, or use target_app_key for an already-known existing app.", "details": {"target_app_ref": ref_value, "available_client_keys": sorted(client_key_indexes.keys())}, } ) elif ref.get("kind") == "target_app": matching_indexes = app_name_indexes.get(ref_value, []) if not matching_indexes: issues.append( { "index": index, "row_number": index + 1, "path": ref.get("path"), "error_code": "TARGET_APP_NAME_UNRESOLVED", "message": f"target_app '{ref_value}' does not match any apps[].app_name in this payload", "fix_hint": "Prefer target_app_ref with a stable apps[].client_key, or make target_app match an app_name exactly.", "details": {"target_app": ref_value, "available_app_names": sorted(app_name_indexes.keys())}, } ) elif len(matching_indexes) > 1: issues.append( { "index": index, "row_number": index + 1, "path": ref.get("path"), "error_code": "TARGET_APP_NAME_AMBIGUOUS", "message": f"target_app '{ref_value}' matches multiple apps[].app_name values", "fix_hint": "Use target_app_ref with a unique apps[].client_key instead of target_app.", "details": {"target_app": ref_value, "matching_indexes": matching_indexes}, } ) if not issues: return None return _config_failure( tool_name=tool_name, error_code="MULTI_APP_STATIC_VALIDATION_FAILED", message="multi-app schema payload has static errors; fix apps[] before writing.", fix_hint="Before creating app shells, ensure each new app has exactly one data title, unique client_key/app_name, and relation refs match apps[].client_key or app_name.", details={"issues": issues, "expected_shape": _schema_apps_expected_shape()}, ) def _compile_multi_app_schema_item_refs( item: JSONObject, client_key_to_app_key: dict[str, str], app_name_to_app_key: dict[str, str] | None = None, ) -> JSONObject: compiled = deepcopy(item) app_name_to_app_key = app_name_to_app_key or {} def visit(value): if isinstance(value, list): return [visit(entry) for entry in value] if not isinstance(value, dict): return value payload = {key: visit(entry) for key, entry in value.items()} ref = ( payload.pop("target_app_ref", None) or payload.pop("targetAppRef", None) or payload.pop("target_app_client_key", None) or payload.pop("targetAppClientKey", None) ) if ref is not None: ref_key = str(ref or "").strip() target_app_key = client_key_to_app_key.get(ref_key) if not target_app_key: raise ValueError(f"target_app_ref '{ref_key}' did not match any apps[].client_key") payload["target_app_key"] = target_app_key target_app = payload.pop("target_app", None) or payload.pop("targetApp", None) if target_app is not None and not str(payload.get("target_app_key") or "").strip(): target_name = str(target_app or "").strip() target_app_key = app_name_to_app_key.get(target_name) if not target_app_key: raise ValueError(f"target_app '{target_name}' did not match any apps[].app_name in the same call") payload["target_app_key"] = target_app_key return payload return visit(compiled) def _split_multi_app_initial_add_fields(item: JSONObject, *, is_new_app: bool) -> tuple[list[JSONObject], list[JSONObject]]: add_fields = _multi_app_list_value(item, "add_fields", "addFields") if not is_new_app: return [], add_fields initial: list[JSONObject] = [] deferred: list[JSONObject] = [] for field in add_fields: if _contains_multi_app_target_ref(field): deferred.append(field) else: initial.append(field) return initial, deferred def _compiled_multi_app_deferred_add_fields(compiled_item: JSONObject, existing_result: JSONObject) -> list[JSONObject]: deferred = existing_result.get("deferred_add_fields") if not isinstance(deferred, list): return list(compiled_item.get("add_fields") or []) deferred_names = {str(item.get("name") or item.get("title") or item.get("label") or "").strip() for item in deferred if isinstance(item, dict)} if not deferred_names: return [] return [ deepcopy(field) for field in list(compiled_item.get("add_fields") or []) if isinstance(field, dict) and str(field.get("name") or field.get("title") or field.get("label") or "").strip() in deferred_names ] def _multi_app_child_diagnostics(result: JSONObject) -> JSONObject: diagnostics: JSONObject = {} for key in ( "backend_code", "http_status", "request_id", "details", "suggested_next_call", "warnings", "publish_result", "duration_breakdown_ms", ): value = result.get(key) if value is None: continue if isinstance(value, (dict, list)) and not value: continue diagnostics[key] = deepcopy(value) return diagnostics def _multi_app_field_type(field: JSONObject) -> str: raw_type = str(field.get("type") or field.get("field_type") or field.get("fieldType") or "").strip() if raw_type: normalized = FIELD_TYPE_ALIASES.get(raw_type.lower()) return normalized.value if normalized is not None else raw_type raw_type_id = field.get("type_id") or field.get("typeId") or field.get("field_type_id") or field.get("fieldTypeId") try: normalized_by_id = FIELD_TYPE_ID_ALIASES.get(int(raw_type_id)) except (TypeError, ValueError): normalized_by_id = None return normalized_by_id.value if normalized_by_id is not None else "" def _multi_app_deferred_relation_retryable(result: JSONObject, deferred_add_fields: list[JSONObject]) -> bool: if not deferred_add_fields: return False if not any(_multi_app_field_type(field) == PublicFieldType.relation.value for field in deferred_add_fields if isinstance(field, dict)): return False if result.get("status") in {"success", "partial_success"}: return False if result.get("error_code") != "SCHEMA_APPLY_FAILED": return False if _schema_apply_result_has_write(result): return False return backend_code_value_int(_result_backend_code(result)) == 400 def _result_backend_code(result: JSONObject) -> Any: if result.get("backend_code") is not None: return result.get("backend_code") details = result.get("details") if isinstance(details, dict): transport = details.get("transport_error") if isinstance(transport, dict): return transport.get("backend_code") return None def _multi_app_relation_target_app_keys(fields: list[JSONObject]) -> list[str]: app_keys: list[str] = [] def visit(value: Any) -> None: if isinstance(value, list): for item in value: visit(item) return if not isinstance(value, dict): return target_app_key = str(value.get("target_app_key") or value.get("targetAppKey") or "").strip() if target_app_key and target_app_key not in app_keys: app_keys.append(target_app_key) for child in value.values(): if isinstance(child, (dict, list)): visit(child) visit(fields) return app_keys def _multi_app_list_value(item: JSONObject, *keys: str) -> list[JSONObject]: for key in keys: value = item.get(key) if isinstance(value, list): return [deepcopy(entry) for entry in value if isinstance(entry, dict)] return [] def _contains_multi_app_target_ref(value: object) -> bool: if isinstance(value, list): return any(_contains_multi_app_target_ref(item) for item in value) if not isinstance(value, dict): return False for key, entry in value.items(): if key in {"target_app_ref", "targetAppRef", "target_app_client_key", "targetAppClientKey", "target_app", "targetApp"}: return True if _contains_multi_app_target_ref(entry): return True return False def _merge_schema_field_diffs(*diffs: object) -> JSONObject: merged: JSONObject = {"added": [], "updated": [], "removed": []} for diff in diffs: if not isinstance(diff, dict): continue for key in ("added", "updated", "removed"): values = diff.get(key) if not isinstance(values, list): continue for value in values: if value not in merged[key]: merged[key].append(value) return merged def _schema_apply_result_has_write(result: JSONObject) -> bool: if "write_executed" in result: return bool(result.get("write_executed")) if bool(result.get("created")) or bool(result.get("published")) or bool(result.get("app_base_updated")): return True field_diff = result.get("field_diff") if isinstance(field_diff, dict): return any(bool(field_diff.get(key)) for key in ("added", "updated", "removed")) return False def _schema_apply_result_may_have_write(result: JSONObject) -> bool: verification = result.get("verification") if isinstance(result.get("verification"), dict) else {} return bool( result.get("write_may_have_succeeded") or str(result.get("next_action") or "") == "readback_before_retry" or verification.get("readback_before_retry") ) def _validation_failure( detail: str, *, tool_name: str | None = None, exc: ValidationError | None = None, suggested_next_call: JSONObject | None = None, ) -> JSONObject: contract = _BUILDER_TOOL_CONTRACTS.get(tool_name or "") reason_path = None if exc is not None: errors = exc.errors() if errors: loc = errors[0].get("loc") if isinstance(loc, (tuple, list)): reason_path = ".".join(str(part) for part in loc) canonical_arguments = None if isinstance(suggested_next_call, dict): arguments = suggested_next_call.get("arguments") if isinstance(arguments, dict): canonical_arguments = arguments return { "status": "failed", "error_code": "VALIDATION_ERROR", "recoverable": True, "message": detail, "normalized_args": {}, "missing_fields": [], "allowed_values": deepcopy(contract.get("allowed_values", {})) if isinstance(contract, dict) else {}, "details": { "validation_detail": detail, "reason_path": reason_path, "allowed_keys": deepcopy(contract.get("allowed_keys", [])) if isinstance(contract, dict) else [], "canonical_arguments": canonical_arguments, "section_allowed_keys": deepcopy(contract.get("section_allowed_keys", [])) if isinstance(contract, dict) else [], "section_aliases": deepcopy(contract.get("section_aliases", {})) if isinstance(contract, dict) else {}, "minimal_section_example": deepcopy(contract.get("minimal_section_example")) if isinstance(contract, dict) else None, }, "suggested_next_call": suggested_next_call, "request_id": None, "backend_code": None, "http_status": None, "noop": False, "verification": {}, } def _visibility_validation_failure( detail: str, *, tool_name: str, exc: ValidationError | None, suggested_next_call: JSONObject | None = None, ) -> JSONObject: result = _validation_failure( detail, tool_name=tool_name, exc=exc, suggested_next_call=suggested_next_call, ) lowered = detail.lower() code_by_fragment = [ ("visibility and auth cannot be provided together", "VISIBILITY_CONFLICT"), ("specific visibility requires selectors", "VISIBILITY_SELECTOR_REQUIRED"), ("selectors are only allowed when mode=specific", "VISIBILITY_SELECTOR_UNEXPECTED"), ("external_mode=specific requires external_selectors", "EXTERNAL_VISIBILITY_SELECTOR_REQUIRED"), ("external_selectors are only allowed when external_mode=specific", "EXTERNAL_VISIBILITY_SELECTOR_UNEXPECTED"), ("mode=everyone requires external_mode=workspace", "VISIBILITY_EXTERNAL_MODE_CONFLICT"), ("external_selectors are not allowed when mode=everyone", "VISIBILITY_SELECTOR_UNEXPECTED"), ] for fragment, error_code in code_by_fragment: if fragment in lowered: result["error_code"] = error_code result["message"] = fragment details = result.get("details") if isinstance(details, dict): details["visibility_error"] = True break return result _RESERVED_SYSTEM_FIELD_NAMES = { "数据ID", "编号", "申请人", "申请时间", "创建人", "创建时间", "提交人", "提交时间", "更新时间", "更新人", "当前流程状态", "当前处理人", "当前处理节点", "流程标题", } def _field_patch_name(item: object) -> str: if not isinstance(item, dict): return "" value = item.get("name") if value is None: value = item.get("title") if value is None: value = item.get("field_name") if value is None: value = item.get("fieldName") return str(value or "").strip() def _reserved_system_field_name_failure( *, tool_name: str, profile: str, app_key: str = "", package_id: int | None = None, app_name: str = "", icon: str | None = None, color: str | None = None, visibility: JSONObject | None = None, publish: bool | None = None, add_fields: list[JSONObject] | None = None, update_fields: list[JSONObject] | None = None, remove_fields: list[JSONObject] | None = None, app_index: int | None = None, ) -> JSONObject | None: for index, item in enumerate(add_fields or []): name = _field_patch_name(item) if name not in _RESERVED_SYSTEM_FIELD_NAMES: continue suggested_add_fields = [field for field in (add_fields or []) if _field_patch_name(field) not in _RESERVED_SYSTEM_FIELD_NAMES] arguments: JSONObject = { "profile": profile, "app_key": app_key, "package_id": package_id, "app_name": app_name, "icon": icon or "", "color": color or "", "visibility": visibility, "add_fields": suggested_add_fields, "update_fields": update_fields or [], "remove_fields": remove_fields or [], } if publish is not None: arguments["publish"] = publish return _config_failure( tool_name=tool_name, error_code="RESERVED_SYSTEM_FIELD_NAME", message=f"add_fields[{index}].name uses built-in system field name '{name}'", fix_hint="Do not create form fields named 数据ID, 编号, 申请人, 申请时间, 创建人, 创建时间, 提交人, 提交时间, 更新时间, 更新人, 当前流程状态, 当前处理人, 当前处理节点, or 流程标题. These fields are provided by Qingflow. Remove them from add_fields; only reference them where the tool explicitly supports system fields, such as button source_field 数据ID.", details={ "reserved_field_names": sorted(_RESERVED_SYSTEM_FIELD_NAMES), "blocked_index": index, "blocked_name": name, "app_index": app_index, }, suggested_next_call={"tool_name": tool_name, "arguments": arguments}, ) return None def _reserved_system_field_name_failure_for_apps( *, tool_name: str, profile: str, package_id: int | None, visibility: JSONObject | None, publish: bool, apps: list[JSONObject], ) -> JSONObject | None: for index, item in enumerate(apps or []): if not isinstance(item, dict): continue failure = _reserved_system_field_name_failure( tool_name=tool_name, profile=profile, app_key=str(item.get("app_key") or item.get("appKey") or ""), package_id=package_id, app_name=str(item.get("app_name") or item.get("appName") or ""), icon=str(item.get("icon") or ""), color=str(item.get("color") or ""), visibility=item.get("visibility") if isinstance(item.get("visibility"), dict) else visibility, publish=publish, add_fields=list(item.get("add_fields") or item.get("addFields") or []), update_fields=list(item.get("update_fields") or item.get("updateFields") or []), remove_fields=list(item.get("remove_fields") or item.get("removeFields") or []), app_index=index, ) if failure is not None: suggested = deepcopy(failure.get("suggested_next_call") or {}) arguments = suggested.get("arguments") if isinstance(arguments, dict): fixed_apps = deepcopy(apps) if isinstance(fixed_apps[index], dict): fixed_apps[index]["add_fields"] = arguments.get("add_fields") or [] arguments = { "profile": profile, "package_id": package_id, "visibility": visibility, "publish": publish, "apps": fixed_apps, } failure["suggested_next_call"] = {"tool_name": tool_name, "arguments": arguments} return failure return None _RESERVED_SYSTEM_VIEW_NAMES = { "全部数据", "我的数据", "我发起的", "待办", "已办", "抄送", } def _reserved_system_view_name_failure( *, tool_name: str, profile: str, app_key: str, publish: bool | None = None, upsert_views: list[JSONObject] | None = None, patch_views: list[JSONObject] | None = None, remove_views: list[str] | None = None, ) -> JSONObject | None: for index, item in enumerate(upsert_views or []): if not isinstance(item, dict): continue name = str(item.get("name") or "").strip() view_key = str(item.get("view_key") or item.get("viewKey") or "").strip() if name in _RESERVED_SYSTEM_VIEW_NAMES and not view_key: suggested: JSONObject = { "tool_name": tool_name, "arguments": { "profile": profile, "app_key": app_key, "upsert_views": [ { **item, "name": "业务台账视图", } ], "patch_views": patch_views or [], "remove_views": remove_views or [], }, } if publish is not None: suggested["arguments"]["publish"] = publish return _config_failure( tool_name=tool_name, error_code="RESERVED_SYSTEM_VIEW_NAME", message=f"upsert_views[{index}].name uses built-in system view name '{name}' without view_key", fix_hint="Do not create business views named 全部数据, 我的数据, 我发起的, 待办, 已办, or 抄送. Use a business-specific view name for new views; to modify a built-in view, pass its existing raw view_key or use patch_views.", details={ "reserved_view_names": sorted(_RESERVED_SYSTEM_VIEW_NAMES), "blocked_index": index, "blocked_name": name, }, suggested_next_call=suggested, ) return None def _view_payload_has_action_buttons( *, upsert_views: list[JSONObject] | None = None, patch_views: list[JSONObject] | None = None, ) -> bool: for item in upsert_views or []: if isinstance(item, dict) and any(key in item for key in ("action_buttons", "actionButtons")): return True for item in patch_views or []: if not isinstance(item, dict): continue raw_set = item.get("set") if isinstance(raw_set, dict) and any(key in raw_set for key in ("action_buttons", "actionButtons")): return True return False def _config_failure( *, tool_name: str, message: str, fix_hint: str, error_code: str = "CONFIG_ERROR", details: JSONObject | None = None, allowed_values: JSONObject | None = None, suggested_next_call: JSONObject | None = None, ) -> JSONObject: contract = _BUILDER_TOOL_CONTRACTS.get(tool_name or "") public_allowed_values = deepcopy(contract.get("allowed_values", {})) if isinstance(contract, dict) else {} if allowed_values: public_allowed_values.update(deepcopy(allowed_values)) public_details: JSONObject = { "fix_hint": fix_hint, "allowed_keys": deepcopy(contract.get("allowed_keys", [])) if isinstance(contract, dict) else [], } if details: public_details.update(deepcopy(details)) return { "status": "failed", "error_code": error_code, "recoverable": True, "message": message, "normalized_args": {}, "missing_fields": [], "allowed_values": public_allowed_values, "details": public_details, "suggested_next_call": suggested_next_call, "request_id": None, "backend_code": None, "http_status": None, "noop": False, "verification": {}, } def _workspace_icon_config_failure( *, tool_name: str, error_code: str, message: str, details: JSONObject, ) -> JSONObject: return _config_failure( tool_name=tool_name, error_code=error_code, message=message, fix_hint="Call `qingflow --json builder icon catalog`, choose an explicit non-template icon and color, then retry.", details=details, allowed_values={ "workspace_icon.icon_names": list(WORKSPACE_ICON_NAMES), "workspace_icon.icon_colors": list(WORKSPACE_ICON_COLORS), "workspace_icon.generic_icon_names": list(GENERIC_WORKSPACE_ICON_NAMES), }, ) def _validate_workspace_icon_for_builder( *, tool_name: str, icon: str | None, color: str | None, creating: bool, ) -> JSONObject | None: if not creating and not (str(icon or "").strip() or str(color or "").strip()): return None ok, error_code, message, details = validate_workspace_icon_choice( icon=icon, color=color, require_explicit=creating, disallow_generic=creating, ) if ok: return None return _workspace_icon_config_failure( tool_name=tool_name, error_code=error_code or "WORKSPACE_ICON_INVALID", message=message or "invalid workspace icon configuration", details=details, ) def _safe_tool_call( call, *, error_code: str, normalized_args: JSONObject, suggested_next_call: JSONObject | None, ) -> JSONObject: started_at = time.perf_counter() try: result = call() except (QingflowApiError, RuntimeError) as error: api_error = _coerce_api_error(error) public_http_status = None if api_error.http_status == 404 else api_error.http_status result = { "status": "failed", "error_code": error_code, "recoverable": True, "message": _public_error_message(error_code, api_error), "normalized_args": normalized_args, "missing_fields": [], "allowed_values": {}, "details": { "transport_error": { "http_status": api_error.http_status, "backend_code": api_error.backend_code, "category": api_error.category, } }, "suggested_next_call": suggested_next_call, "request_id": api_error.request_id, "backend_code": api_error.backend_code, "http_status": public_http_status, "noop": False, "verification": {}, } except ValueError as error: public_error_code, public_message = _public_value_error(error, fallback_error_code=error_code) result = { "status": "failed", "error_code": public_error_code, "recoverable": True, "message": public_message, "normalized_args": normalized_args, "missing_fields": [], "allowed_values": {}, "details": { "validation_error": { "message": str(error), "fallback_error_code": error_code, } }, "suggested_next_call": suggested_next_call, "write_executed": False, "safe_to_retry": True, "noop": False, "verification": {}, } if isinstance(result, dict): _merge_duration_breakdown(result, tool_call_ms=_elapsed_ms(started_at)) return result def _public_value_error(error: ValueError, *, fallback_error_code: str) -> tuple[str, str]: message = str(error).strip() or fallback_error_code prefix, separator, rest = message.partition(":") candidate = prefix.strip() if separator and _looks_like_public_error_code(candidate): return candidate, rest.strip() or message return fallback_error_code, message def _looks_like_public_error_code(value: str) -> bool: if not value or len(value) > 80 or not any(ch.isalpha() for ch in value): return False return all(ch.isupper() or ch.isdigit() or ch == "_" for ch in value) def _publicize_package_fields(value): if isinstance(value, list): return [_publicize_package_fields(item) for item in value] if not isinstance(value, dict): return value key_map = { "tag_id": "package_id", "tag_ids": "package_ids", "tag_ids_before": "package_ids_before", "tag_ids_after": "package_ids_after", "tag_name": "package_name", "tag_icon": "icon", "iconConfig": "icon_config", "package_tag_id": "package_id", "package_tag_ids": "package_ids", "expected_package_tag_id": "expected_package_id", } public: JSONObject = {} for key, item in value.items(): public_key = key_map.get(key, key) public[public_key] = _publicize_package_fields(item) return public def _builder_contract_with_apply_output(tool_name: str, contract: JSONObject) -> JSONObject: public = deepcopy(contract) if tool_name not in BUILDER_APPLY_TOOL_NAMES: return public notes = public.setdefault("execution_notes", []) if isinstance(notes, list): note = "apply/write output includes schema_version, operation, summary, and resources[]; UI and agents should read resources[].id/key/name first and use legacy fields only for compatibility/debugging" if note not in notes: notes.append(note) public["output_contract"] = { "schema_version": BUILDER_APPLY_SCHEMA_VERSION, "preferred_ui_fields": ["operation", "summary", "resources"], "resource_fields": ["resource_type", "operation", "status", "id", "key", "name", "ids", "parent", "icon_config", "error_code", "message"], "json_paths": _builder_apply_json_paths(), "legacy_fields_preserved": True, } return public def _builder_tool_contract_summary(tool_name: str, contract: JSONObject) -> JSONObject: allowed_values = contract.get("allowed_values") if isinstance(contract, dict) else {} allowed_values_keys = sorted(str(key) for key in allowed_values) if isinstance(allowed_values, dict) else [] return { "tool_name": tool_name, "contract_path": "$.contract", "allowed_keys_path": "$.contract.allowed_keys", "allowed_values_path": "$.contract.allowed_values", "allowed_values_key_style": "flat_dotted_keys", "minimal_example_path": "$.contract.minimal_example", "execution_notes_path": "$.contract.execution_notes", "top_level_allowed_values_usage": "empty on successful contract lookup; use $.contract.allowed_values instead", "allowed_values_keys_sample": allowed_values_keys[:12], "json_paths": { "contract": "$.contract", "allowed_keys": "$.contract.allowed_keys", "allowed_values": "$.contract.allowed_values", "minimal_example": "$.contract.minimal_example", "execution_notes": "$.contract.execution_notes", "aliases": "$.contract.aliases", }, } def _builder_apply_json_paths() -> JSONObject: return { "status": "$.status", "schema_version": "$.schema_version", "operation": "$.operation", "summary": "$.summary", "resources": "$.resources", "warnings": "$.warnings", "verification": "$.verification", "details": "$.details", "duration_breakdown_ms": "$.duration_breakdown_ms", "total_ms": "$.duration_breakdown_ms.total_ms", "write_executed": "$.write_executed", "write_may_have_succeeded": "$.write_may_have_succeeded", "safe_to_retry": "$.safe_to_retry", "next_action": "$.next_action", "summary_write_executed": "$.summary.write_executed", "summary_safe_to_retry": "$.summary.safe_to_retry", "resource_keys": "$.resources[].key", "resource_ids": "$.resources[].id", "resource_statuses": "$.resources[].status", } def _attach_builder_apply_envelope(tool_name: str, payload: JSONObject) -> JSONObject: if not isinstance(payload, dict): return payload resources = _builder_apply_resources(tool_name, payload) payload["schema_version"] = BUILDER_APPLY_SCHEMA_VERSION payload["operation"] = tool_name payload["resources"] = resources payload["summary"] = _builder_apply_summary(payload, resources) payload["json_paths"] = _builder_apply_json_paths() return payload def _builder_apply_summary(payload: JSONObject, resources: list[JSONObject]) -> JSONObject: status = str(payload.get("status") or "") failed = sum(1 for item in resources if str(item.get("status") or "") == "failed") created = sum(1 for item in resources if str(item.get("operation") or "") == "created" and str(item.get("status") or "") != "failed") removed = sum(1 for item in resources if str(item.get("operation") or "") == "removed" and str(item.get("status") or "") != "failed") updated = sum( 1 for item in resources if str(item.get("status") or "") != "failed" and str(item.get("operation") or "") in {"updated", "layout_updated", "workflow_updated", "verified", "published"} ) published_value = payload.get("published") if published_value is None: published_value = _builder_batch_published_value(payload) if published_value is None: publish_requested = payload.get("publish_requested") published_value = bool(publish_requested) and status in {"success", "partial_success"} verified_value = payload.get("verified") if verified_value is None: verification = payload.get("verification") verified_value = status == "success" and failed == 0 and _builder_verification_truthy(verification) summary: JSONObject = { "total": len(resources), "created": created, "updated": updated, "removed": removed, "failed": failed, "published": bool(published_value), "verified": bool(verified_value), } if "write_executed" in payload: summary["write_executed"] = bool(payload.get("write_executed")) if "write_may_have_succeeded" in payload: summary["write_may_have_succeeded"] = bool(payload.get("write_may_have_succeeded")) if "safe_to_retry" in payload: summary["safe_to_retry"] = bool(payload.get("safe_to_retry")) if payload.get("next_action"): summary["next_action"] = payload.get("next_action") if payload.get("pending_readback") is not None: summary["pending_readback"] = payload.get("pending_readback") return summary def _builder_batch_published_value(payload: JSONObject) -> bool | None: apps = payload.get("apps") if not isinstance(apps, list) or not apps: return None values: list[bool] = [] for item in apps: if not isinstance(item, dict): continue result = item.get("result") if isinstance(item.get("result"), dict) else {} if "published" in result: values.append(bool(result.get("published"))) elif "published" in item: values.append(bool(item.get("published"))) if not values: return None return all(values) def _builder_verification_truthy(value: object) -> bool: if value is None: return True if isinstance(value, bool): return value if isinstance(value, dict): booleans = [item for item in value.values() if isinstance(item, bool)] return all(booleans) if booleans else True return True def _builder_apply_resources(tool_name: str, payload: JSONObject) -> list[JSONObject]: resources: list[JSONObject] if tool_name == "package_apply": resources = _builder_package_resources(payload) elif tool_name == "app_schema_apply": resources = _builder_schema_resources(payload) elif isinstance(payload.get("apps"), list) and _builder_apply_tool_is_app_scoped(tool_name): resources = _builder_batch_app_resources(tool_name, payload) elif tool_name == "app_layout_apply": resources = [_builder_app_resource(payload, operation="layout_updated")] elif tool_name == "app_flow_apply": resources = [_builder_app_resource(payload, operation="workflow_updated")] elif tool_name == "app_views_apply": resources = _builder_view_resources(payload) elif tool_name == "app_charts_apply": resources = _builder_chart_resources(payload) elif tool_name == "portal_apply": resources = _builder_portal_resources(payload) elif tool_name == "portal_delete": resources = _builder_portal_resources(payload, operation_override="removed") elif tool_name == "app_custom_buttons_apply": resources = _builder_button_resources(payload) elif tool_name == "app_associated_resources_apply": resources = _builder_associated_resource_resources(payload) elif tool_name == "app_publish_verify": resources = [_builder_app_resource(payload, operation="verified")] else: resources = [] if not resources and _builder_status(payload, "") == "failed" and _builder_apply_tool_is_app_scoped(tool_name): app_key = _builder_payload_app_key(payload) if app_key not in (None, ""): resources = [_builder_app_resource(payload, operation="failed")] return resources def _builder_batch_app_resources(tool_name: str, payload: JSONObject) -> list[JSONObject]: resources: list[JSONObject] = [] operation_by_tool = { "app_schema_apply": "updated", "app_layout_apply": "layout_updated", "app_flow_apply": "workflow_updated", "app_views_apply": "updated", "app_custom_buttons_apply": "updated", "app_associated_resources_apply": "updated", "app_charts_apply": "updated", "app_publish_verify": "verified", } operation = operation_by_tool.get(tool_name, "updated") for item in payload.get("apps") or []: if not isinstance(item, dict): continue result = item.get("result") if isinstance(item.get("result"), dict) else {} nested_resources = result.get("resources") if isinstance(result, dict) else None if isinstance(nested_resources, list) and nested_resources: resources.extend(resource for resource in nested_resources if isinstance(resource, dict)) continue app_payload: JSONObject = { **(result if isinstance(result, dict) else {}), "app_key": item.get("app_key") or (result.get("app_key") if isinstance(result, dict) else None), "status": item.get("status") or (result.get("status") if isinstance(result, dict) else None), "error_code": item.get("error_code") or (result.get("error_code") if isinstance(result, dict) else None), "message": item.get("message") or (result.get("message") if isinstance(result, dict) else None), } resources.append(_builder_app_resource(app_payload, operation=operation)) return resources def _builder_apply_tool_is_app_scoped(tool_name: str) -> bool: return tool_name in { "app_schema_apply", "app_layout_apply", "app_flow_apply", "app_views_apply", "app_custom_buttons_apply", "app_associated_resources_apply", "app_charts_apply", "app_publish_verify", } def _builder_status(payload_or_item: JSONObject, fallback: str = "success") -> str: status = str(payload_or_item.get("status") or fallback or "success") return status def _builder_operation(value: object, fallback: str = "updated") -> str: raw = str(value or fallback or "updated").strip() mapping = { "create": "created", "created": "created", "add": "created", "update": "updated", "updated": "updated", "patch": "updated", "remove": "removed", "removed": "removed", "delete": "removed", "deleted": "removed", "unchanged": "unchanged", "failed": "failed", } return mapping.get(raw, raw) def _builder_parent(resource_type: str, *, key: object = None, name: object = None, id_value: object = None) -> JSONObject: return { "resource_type": resource_type, "id": id_value, "key": str(key) if key not in (None, "") else None, "name": str(name) if name not in (None, "") else None, } def _builder_app_parent(payload: JSONObject) -> JSONObject | None: app_key = payload.get("app_key") or payload.get("appKey") app_name = payload.get("app_name_after") or payload.get("app_name") or payload.get("appTitle") or payload.get("app_title") if app_key in (None, "") and app_name in (None, ""): return None return _builder_parent("app", key=app_key, name=app_name) def _builder_icon_config(raw_icon: object = None, *, icon: object = None, color: object = None) -> JSONObject | None: raw = str(raw_icon).strip() if raw_icon not in (None, "") else "" explicit_icon = str(icon).strip() if icon not in (None, "") else "" explicit_color = str(color).strip() if color not in (None, "") else "" if raw: if raw.startswith("{") and raw.endswith("}"): config = workspace_icon_config(raw) else: config = { "icon_name": normalize_workspace_icon_name(raw), "icon_color": explicit_color or None, "icon_text": None, "raw": raw, } if any(config.get(key) for key in ("icon_name", "icon_color", "icon_text", "raw")): return config if explicit_icon or explicit_color: return { "icon_name": normalize_workspace_icon_name(explicit_icon) if explicit_icon else None, "icon_color": explicit_color or None, "icon_text": None, "raw": None, } return None def _builder_container_icon_config(container: object, *, raw_keys: tuple[str, ...], icon_keys: tuple[str, ...] = ("icon",), color_keys: tuple[str, ...] = ("color",)) -> JSONObject | None: if not isinstance(container, dict): return None raw_icon = next((container.get(key) for key in raw_keys if container.get(key) not in (None, "")), None) icon = next((container.get(key) for key in icon_keys if container.get(key) not in (None, "")), None) color = next((container.get(key) for key in color_keys if container.get(key) not in (None, "")), None) return _builder_icon_config(raw_icon, icon=icon, color=color) def _builder_resource( *, resource_type: str, operation: str, status: str, id_value: object = None, key: object = None, name: object = None, ids: JSONObject | None = None, parent: JSONObject | None = None, icon_config: JSONObject | None = None, error_code: object = None, message: object = None, ) -> JSONObject: resource = { "resource_type": resource_type, "operation": operation, "status": status, "id": id_value, "key": str(key) if key not in (None, "") else None, "name": str(name) if name not in (None, "") else None, "ids": ids or {}, "parent": parent, "error_code": str(error_code) if error_code not in (None, "") else None, "message": str(message) if message not in (None, "") else None, } if icon_config: resource["icon_config"] = icon_config return resource def _builder_app_resource(payload: JSONObject, *, operation: str) -> JSONObject: status = _builder_status(payload, "success") if status == "failed": operation = "failed" app_key = _builder_payload_app_key(payload) app_name = _builder_payload_app_name(payload) normalized_args = payload.get("normalized_args") if isinstance(payload.get("normalized_args"), dict) else {} icon_config = ( _builder_container_icon_config(payload, raw_keys=("app_icon", "appIcon")) or _builder_container_icon_config(normalized_args, raw_keys=("app_icon", "appIcon", "icon")) ) return _builder_resource( resource_type="app", operation=operation, status=status, key=app_key, name=app_name, ids={"app_key": app_key} if app_key not in (None, "") else {}, icon_config=icon_config, error_code=payload.get("error_code"), message=payload.get("message") if status == "failed" else None, ) def _builder_payload_app_key(payload: JSONObject) -> object: return _builder_payload_identity_value(payload, ("app_key", "appKey")) def _builder_payload_app_name(payload: JSONObject) -> object: return _builder_payload_identity_value(payload, ("app_name_after", "app_name", "appName", "appTitle", "app_title", "name", "title")) def _builder_payload_identity_value(payload: JSONObject, keys: tuple[str, ...]) -> object: for key in keys: value = payload.get(key) if value not in (None, ""): return value for container_key in ("normalized_args", "canonical_arguments"): container = payload.get(container_key) if isinstance(container, dict): for key in keys: value = container.get(key) if value not in (None, ""): return value details = payload.get("details") if isinstance(details, dict): for container_key in ("normalized_args", "canonical_arguments"): container = details.get(container_key) if isinstance(container, dict): for key in keys: value = container.get(key) if value not in (None, ""): return value return None def _builder_package_resources(payload: JSONObject) -> list[JSONObject]: package_id = payload.get("package_id") or payload.get("id") package_name = payload.get("package_name") or payload.get("name") status = _builder_status(payload, "success") operation = ( "failed" if status == "failed" else "removed" if bool(payload.get("deleted")) or bool(payload.get("delete_executed")) else "created" if bool(payload.get("created")) else "updated" ) normalized_args = payload.get("normalized_args") if isinstance(payload.get("normalized_args"), dict) else {} icon_config = ( _builder_container_icon_config(payload, raw_keys=("icon", "tagIcon", "tag_icon")) or _builder_container_icon_config(normalized_args, raw_keys=("icon", "tagIcon", "tag_icon")) ) return [ _builder_resource( resource_type="package", operation=operation, status=status, id_value=package_id, key=str(package_id) if package_id not in (None, "") else None, name=package_name, ids={"package_id": package_id} if package_id not in (None, "") else {}, icon_config=icon_config, error_code=payload.get("error_code"), message=payload.get("message") if status == "failed" else None, ) ] def _builder_schema_resources(payload: JSONObject) -> list[JSONObject]: if payload.get("mode") == "multi_app" and isinstance(payload.get("apps"), list): resources: list[JSONObject] = [] package_id = payload.get("package_id") package_parent = _builder_parent("package", id_value=package_id, key=package_id) if package_id not in (None, "") else None for item in payload.get("apps") or []: if not isinstance(item, dict): continue status = _builder_status(item, "success") operation = "failed" if status == "failed" else ("created" if bool(item.get("created")) else "updated") parent = _builder_parent("app", key=item.get("app_key"), name=item.get("app_name")) icon_config = ( _builder_container_icon_config(item, raw_keys=("app_icon", "appIcon", "icon")) or _builder_container_icon_config(item.get("shell_result"), raw_keys=("app_icon", "appIcon", "icon")) ) resources.append( _builder_resource( resource_type="app", operation=operation, status=status, key=item.get("app_key"), name=item.get("app_name"), ids={ **({"app_key": item.get("app_key")} if item.get("app_key") else {}), **({"package_id": package_id} if package_id not in (None, "") else {}), }, parent=package_parent, icon_config=icon_config, error_code=item.get("error_code"), message=item.get("message") if status in {"failed", "pending_readback"} else None, ) ) resources.extend(_builder_field_resources(item.get("field_diff_details") or item.get("field_diff"), parent=parent)) return resources status = _builder_status(payload, "success") operation = "failed" if status == "failed" else ("created" if bool(payload.get("created")) else "updated") app_key = payload.get("app_key") app_name = payload.get("app_name_after") or payload.get("app_name") parent = _builder_parent("app", key=app_key, name=app_name) normalized_args = payload.get("normalized_args") if isinstance(payload.get("normalized_args"), dict) else {} icon_config = ( _builder_container_icon_config(payload, raw_keys=("app_icon", "appIcon", "icon")) or _builder_container_icon_config(normalized_args, raw_keys=("icon", "app_icon", "appIcon")) ) resources = [ _builder_resource( resource_type="app", operation=operation, status=status, key=app_key, name=app_name, ids={"app_key": app_key} if app_key else {}, icon_config=icon_config, error_code=payload.get("error_code"), message=payload.get("message") if status == "failed" else None, ) ] resources.extend(_builder_field_resources(payload.get("field_diff_details") or payload.get("field_diff"), parent=parent)) return resources def _builder_field_resources(field_diff: object, *, parent: JSONObject | None) -> list[JSONObject]: if not isinstance(field_diff, dict): return [] resources: list[JSONObject] = [] for key, operation in (("added", "created"), ("updated", "updated"), ("removed", "removed")): for field in field_diff.get(key) or []: if isinstance(field, dict): name = field.get("name") or field.get("title") or field.get("field_name") or field.get("queTitle") field_id = field.get("field_id") or field.get("queId") que_id = field.get("que_id") or field.get("queId") else: name = field field_id = None que_id = None resource_id = que_id if que_id not in (None, "") else field_id resources.append( _builder_resource( resource_type="field", operation=operation, status="success", id_value=resource_id, key=field_id if field_id not in (None, "") else name, name=name, ids={ **({"field_id": field_id} if field_id not in (None, "") else {}), **({"que_id": que_id} if que_id not in (None, "") else {}), **({"app_key": parent.get("key")} if isinstance(parent, dict) and parent.get("key") else {}), }, parent=parent, ) ) return resources def _builder_view_resources(payload: JSONObject) -> list[JSONObject]: if isinstance(payload.get("views"), list): resources: list[JSONObject] = [] for item in payload.get("views") or []: if not isinstance(item, dict): continue result = item.get("result") if isinstance(item.get("result"), dict) else {} parent = _builder_app_parent({"app_key": item.get("app_key") or result.get("app_key")}) operation = str(item.get("operation") or item.get("requested_operation") or "updated") if operation == "upsert": operation = "updated" resources.append( _builder_resource( resource_type="view", operation=operation, status=str(item.get("status") or result.get("status") or "success"), key=item.get("view_key") or item.get("viewKey"), name=item.get("name") or item.get("view_name") or item.get("viewName"), ids={ **({"view_key": item.get("view_key") or item.get("viewKey")} if item.get("view_key") or item.get("viewKey") else {}), **({"app_key": item.get("app_key")} if item.get("app_key") else {}), }, parent=parent, error_code=item.get("error_code") or result.get("error_code"), message=item.get("message") or result.get("message"), ) ) return resources diff = payload.get("views_diff") if isinstance(payload.get("views_diff"), dict) else {} verification_by_name = _builder_view_verification_by_name(payload) parent = _builder_app_parent(payload) resources: list[JSONObject] = [] for key, operation in (("created", "created"), ("updated", "updated"), ("removed", "removed")): for item in diff.get(key) or []: name, view_key, status, error_code, message = _builder_view_identity(item, verification_by_name) resources.append( _builder_resource( resource_type="view", operation=operation, status=status, key=view_key, name=name, ids={ **({"view_key": view_key} if view_key else {}), **({"app_key": payload.get("app_key")} if payload.get("app_key") else {}), }, parent=parent, error_code=error_code, message=message, ) ) for item in diff.get("failed") or []: name, view_key, _status, error_code, message = _builder_view_identity(item, verification_by_name) resources.append( _builder_resource( resource_type="view", operation="failed", status="failed", key=view_key, name=name, ids={ **({"view_key": view_key} if view_key else {}), **({"app_key": payload.get("app_key")} if payload.get("app_key") else {}), }, parent=parent, error_code=error_code, message=message, ) ) details = payload.get("details") if isinstance(payload.get("details"), dict) else {} action_buttons_result = details.get("action_buttons_result") if isinstance(details, dict) else None if isinstance(action_buttons_result, dict): nested_payload = { **action_buttons_result, "app_key": payload.get("app_key"), "app_name": payload.get("app_name"), } resources.extend(_builder_button_resources(nested_payload)) return resources def _builder_view_verification_by_name(payload: JSONObject) -> dict[str, JSONObject]: verification = payload.get("verification") by_view = verification.get("by_view") if isinstance(verification, dict) else None result: dict[str, JSONObject] = {} if isinstance(by_view, list): for item in by_view: if isinstance(item, dict): name = str(item.get("name") or "").strip() if name: result[name] = item return result def _builder_view_identity(item: object, verification_by_name: dict[str, JSONObject]) -> tuple[str | None, str | None, str, object, object]: verification: JSONObject | None = None if isinstance(item, dict): name = item.get("name") or item.get("view_name") or item.get("viewName") view_key = item.get("view_key") or item.get("viewKey") status = str(item.get("status") or "success") error_code = item.get("error_code") message = item.get("message") else: name = str(item) if item not in (None, "") else None view_key = None status = "success" error_code = None message = None if name and not view_key: verification = verification_by_name.get(str(name)) if isinstance(verification, dict): view_key = verification.get("view_key") or verification.get("viewKey") if not view_key: matching = verification.get("matching_view_keys") if isinstance(matching, list) and matching: view_key = matching[0] if name and verification is None: verification = verification_by_name.get(str(name)) if isinstance(verification, dict): verification_status = str(verification.get("status") or "").strip() if verification_status in {"removed", "readback_pending"}: status = "readback_pending" if verification_status == "removed": status = "removed" error_code = error_code or verification.get("error_code") message = message or verification.get("message") return ( str(name) if name not in (None, "") else None, str(view_key) if view_key not in (None, "") else None, status, error_code, message, ) def _builder_chart_resources(payload: JSONObject) -> list[JSONObject]: parent = _builder_app_parent(payload) resources: list[JSONObject] = [] for item in payload.get("chart_results") or []: if not isinstance(item, dict): continue status = str(item.get("status") or "success") operation = _builder_operation(item.get("operation") or status, fallback="updated") if status == "failed": operation = "failed" resource_status = "failed" if status == "failed" else ("readback_pending" if status == "readback_pending" else "success") chart_id = item.get("chart_id") or item.get("chartId") chart_key = item.get("chart_key") or item.get("chartKey") resources.append( _builder_resource( resource_type="chart", operation=operation, status=resource_status, id_value=chart_id, key=chart_key or chart_id, name=item.get("name") or item.get("chart_name") or item.get("chartName"), ids={ **({"chart_id": chart_id} if chart_id not in (None, "") else {}), **({"chart_key": chart_key} if chart_key else {}), **({"app_key": payload.get("app_key")} if payload.get("app_key") else {}), **({"chart_type": item.get("chart_type")} if item.get("chart_type") else {}), }, parent=parent, error_code=item.get("error_code"), message=item.get("message") if status in {"failed", "readback_pending"} else None, ) ) return resources def _builder_portal_resources(payload: JSONObject, *, operation_override: str | None = None) -> list[JSONObject]: status = _builder_status(payload, "success") draft_result = payload.get("draft_result") if isinstance(payload.get("draft_result"), dict) else {} live_result = payload.get("live_result") if isinstance(payload.get("live_result"), dict) else {} normalized_args = payload.get("normalized_args") if isinstance(payload.get("normalized_args"), dict) else {} dash_key = ( payload.get("dash_key") or payload.get("dashKey") or draft_result.get("dashKey") or draft_result.get("dash_key") or live_result.get("dashKey") or live_result.get("dash_key") ) dash_name = ( payload.get("dash_name") or payload.get("dashName") or payload.get("name") or draft_result.get("dashName") or draft_result.get("dash_name") or draft_result.get("name") or live_result.get("dashName") or live_result.get("dash_name") or live_result.get("name") or normalized_args.get("dash_name") or normalized_args.get("dashName") ) package_id = payload.get("package_id") or normalized_args.get("package_id") or normalized_args.get("package_tag_id") if package_id in (None, "") and isinstance(draft_result.get("tags"), list) and draft_result.get("tags"): first_tag = draft_result.get("tags")[0] if isinstance(first_tag, dict): package_id = first_tag.get("tagId") or first_tag.get("tag_id") or first_tag.get("id") operation = operation_override or ("failed" if status == "failed" else ("created" if bool(payload.get("created")) else "updated")) parent = None if package_id: parent = _builder_parent("package", id_value=package_id, key=package_id) icon_config = ( _builder_container_icon_config(payload, raw_keys=("dash_icon", "dashIcon", "icon")) or _builder_container_icon_config(draft_result, raw_keys=("dashIcon", "dash_icon", "icon")) or _builder_container_icon_config(live_result, raw_keys=("dashIcon", "dash_icon", "icon")) or _builder_container_icon_config(normalized_args, raw_keys=("icon", "dash_icon", "dashIcon")) ) return [ _builder_resource( resource_type="portal", operation=operation, status=status, key=dash_key, name=dash_name, ids={ **({"dash_key": dash_key} if dash_key else {}), **({"package_id": package_id} if package_id else {}), }, parent=parent, icon_config=icon_config, error_code=payload.get("error_code"), message=payload.get("message") if status == "failed" else None, ) ] def _builder_button_resources(payload: JSONObject) -> list[JSONObject]: parent = _builder_app_parent(payload) resources: list[JSONObject] = [] for key, operation in (("created", "created"), ("updated", "updated"), ("removed", "removed"), ("failed", "failed")): for item in payload.get(key) or []: if not isinstance(item, dict): continue status = "failed" if operation == "failed" or item.get("status") == "failed" else str(item.get("status") or "success") button_id = item.get("button_id") or item.get("buttonId") resources.append( _builder_resource( resource_type="button", operation=operation if operation != "failed" else "failed", status=status, id_value=button_id, key=button_id, name=item.get("button_text") or item.get("buttonText"), ids={ **({"button_id": button_id} if button_id not in (None, "") else {}), **({"app_key": payload.get("app_key")} if payload.get("app_key") else {}), }, parent=parent, error_code=item.get("error_code"), message=item.get("message") if status in {"failed", "readback_pending"} else None, ) ) for item in payload.get("view_configs") or []: if isinstance(item, dict): status = str(item.get("status") or "success") view_key = item.get("view_key") or item.get("viewKey") resources.append( _builder_resource( resource_type="button_binding", operation="updated" if status != "failed" else "failed", status=status, key=view_key, name=item.get("view_name") or item.get("viewName") or view_key, ids={ **({"view_key": view_key} if view_key else {}), **({"app_key": payload.get("app_key")} if payload.get("app_key") else {}), }, parent=parent, error_code=item.get("error_code"), message=item.get("message") if status in {"failed", "readback_pending"} else None, ) ) return resources def _builder_associated_resource_resources(payload: JSONObject) -> list[JSONObject]: parent = _builder_app_parent(payload) readback_by_id = _builder_associated_resource_readback_by_id(payload) resources: list[JSONObject] = [] for key, operation in ( ("created", "created"), ("updated", "updated"), ("unchanged", "unchanged"), ("removed", "removed"), ("failed", "failed"), ): for item in payload.get(key) or []: if not isinstance(item, dict): continue status = "failed" if operation == "failed" or item.get("status") == "failed" else str(item.get("status") or "success") associated_item_id = item.get("associated_item_id") or item.get("associatedItemId") readback = readback_by_id.get(str(associated_item_id)) if associated_item_id not in (None, "") else None view_key = item.get("view_key") or item.get("viewKey") or (readback or {}).get("view_key") or (readback or {}).get("viewKey") chart_key = item.get("chart_key") or item.get("chartKey") or (readback or {}).get("chart_key") or (readback or {}).get("chartKey") target_app_key = item.get("target_app_key") or (readback or {}).get("target_app_key") name = item.get("name") or item.get("resource_name") or (readback or {}).get("name") or view_key or chart_key resources.append( _builder_resource( resource_type="associated_resource", operation=operation if operation != "failed" else "failed", status=status, id_value=associated_item_id, key=view_key or chart_key or associated_item_id, name=name, ids={ **({"associated_item_id": associated_item_id} if associated_item_id not in (None, "") else {}), **({"app_key": payload.get("app_key")} if payload.get("app_key") else {}), **({"target_app_key": target_app_key} if target_app_key else {}), **({"view_key": view_key} if view_key else {}), **({"chart_key": chart_key} if chart_key else {}), }, parent=parent, error_code=item.get("error_code"), message=item.get("message") if status in {"failed", "readback_pending"} else None, ) ) for item in payload.get("view_configs") or []: if isinstance(item, dict): status = str(item.get("status") or "success") view_key = item.get("view_key") or item.get("viewKey") resources.append( _builder_resource( resource_type="associated_resource_binding", operation="updated" if status != "failed" else "failed", status=status, key=view_key, name=item.get("view_name") or item.get("viewName") or view_key, ids={ **({"view_key": view_key} if view_key else {}), **({"app_key": payload.get("app_key")} if payload.get("app_key") else {}), }, parent=parent, error_code=item.get("error_code"), message=item.get("message") if status == "failed" else None, ) ) return resources def _builder_associated_resource_readback_by_id(payload: JSONObject) -> dict[str, JSONObject]: result: dict[str, JSONObject] = {} def collect(items: object) -> None: if not isinstance(items, list): return for item in items: if not isinstance(item, dict): continue associated_item_id = item.get("associated_item_id") or item.get("associatedItemId") if associated_item_id in (None, ""): continue result[str(associated_item_id)] = item collect(payload.get("associated_resources")) for config in payload.get("view_configs") or []: if not isinstance(config, dict): continue for key in ("actual", "expected"): value = config.get(key) if isinstance(value, dict): collect(value.get("items")) return result def _coerce_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 _public_error_message(error_code: str, error: QingflowApiError) -> str: if backend_code_int(error) == 40074 or error_code == "APP_EDIT_LOCKED": return "app is currently locked by another active editor session" if error.http_status != 404: return error.message mapping = { "PACKAGE_LIST_FAILED": "package list is unavailable in the current route", "PACKAGE_RESOLVE_FAILED": "package resolution is unavailable in the current route", "SOLUTION_INSTALL_FAILED": "solution install could not complete because the app creation route or solution source is unavailable", "PACKAGE_ATTACH_FAILED": "package attachment could not be verified in the current route", "APP_RESOLVE_FAILED": "app resolution is unavailable in the current route", "CUSTOM_BUTTON_LIST_FAILED": "custom button list is unavailable in the current route", "CUSTOM_BUTTON_GET_FAILED": "custom button detail is unavailable in the current route", "CUSTOM_BUTTON_CREATE_FAILED": "custom button create could not complete because the route or readback is unavailable", "CUSTOM_BUTTON_UPDATE_FAILED": "custom button update could not complete because the route or readback is unavailable", "CUSTOM_BUTTON_DELETE_FAILED": "custom button delete could not complete because the route is unavailable", "APP_READ_FAILED": "app base or schema is unavailable in the current route", "FIELDS_READ_FAILED": "app fields are unavailable in the current route", "LAYOUT_READ_FAILED": "layout resource is unavailable for this app in the current route", "VIEWS_READ_FAILED": "views resource is unavailable for this app in the current route", "FLOW_READ_FAILED": "workflow resource is unavailable for this app in the current route", "SCHEMA_PLAN_FAILED": "schema planning could not load the required app state in the current route", "LAYOUT_PLAN_FAILED": "layout planning could not load the required app state in the current route", "FLOW_PLAN_FAILED": "flow planning could not load the required app state in the current route", "VIEWS_PLAN_FAILED": "views planning could not load the required app state in the current route", "SCHEMA_APPLY_FAILED": "schema apply could not complete because the app route or readback is unavailable", "LAYOUT_APPLY_FAILED": "layout apply could not complete because the layout route or readback is unavailable", "FLOW_APPLY_FAILED": "flow apply could not complete because the workflow route or readback is unavailable", "VIEWS_APPLY_FAILED": "views apply could not complete because the views route or readback is unavailable", "CHART_APPLY_FAILED": "chart apply could not complete because the QingBI route or readback is unavailable", "PORTAL_APPLY_FAILED": "portal apply could not complete because the portal route or readback is unavailable", "PUBLISH_VERIFY_FAILED": "publish verification is unavailable in the current route", } return mapping.get(error_code, "requested builder resource is unavailable in the current route") _VISIBILITY_ALLOWED_VALUES: JSONObject = { "visibility.mode": ["workspace", "everyone", "specific"], "visibility.external_mode": ["not", "workspace", "specific"], "visibility.selectors": [ "member_uids", "member_emails", "member_names", "dept_ids", "dept_names", "role_ids", "role_names", "include_sub_departs", ], "visibility.external_selectors": ["member_ids", "member_emails", "dept_ids"], } _VISIBILITY_EXECUTION_NOTES = [ "visibility is the canonical public auth shape for packages, apps, views, charts, and portals", "mode=workspace means internal workspace members; mode=everyone means all users; mode=specific requires selectors", "external_mode defaults to not and controls external-user visibility independently unless mode=everyone", "name selectors must resolve to exactly one member, department, or role; ambiguous names fail instead of guessing", "when updating an existing resource, omit visibility to preserve the current backend auth", ] _VISIBILITY_WORKSPACE_EXAMPLE: JSONObject = { "mode": "workspace", "selectors": {}, "external_mode": "not", "external_selectors": {}, } _VISIBILITY_SPECIFIC_EXAMPLE: JSONObject = { "mode": "specific", "selectors": { "member_emails": ["owner@example.com"], "role_names": ["项目经理"], "include_sub_departs": False, }, "external_mode": "not", "external_selectors": {}, } _BUILDER_TOOL_CONTRACTS: dict[str, JSONObject] = { "file_upload_local": { "allowed_keys": ["upload_kind", "file_path", "upload_mark", "content_type", "bucket_type", "path_id", "file_related_url"], "aliases": {}, "allowed_values": {"upload_kind": ["attachment", "login"]}, "execution_notes": [ "returns upload metadata plus attachment-ready values for later builder or user writes", "upload_kind=attachment may fall back to login upload info when the backend rejects attachment upload info", "upload_mark is recommended for attachment uploads because it affects the remote storage path", ], "minimal_example": { "profile": "default", "upload_kind": "attachment", "file_path": "/absolute/path/to/local-file.txt", "upload_mark": "APP_KEY", }, }, "feedback_submit": { "allowed_keys": [ "category", "title", "description", "expected_behavior", "actual_behavior", "impact_scope", "tool_name", "app_key", "record_id", "workflow_node_id", "note", ], "aliases": {}, "allowed_values": {"category": ["bug", "missing_capability", "awkward_workflow", "ux", "docs"]}, "execution_notes": [ "submit feedback only after explicit user confirmation", "use this when the current builder capability is unsupported or awkward after reasonable attempts", "requires feedback qsource configuration but does not require Qingflow login or workspace selection", ], "minimal_example": { "category": "missing_capability", "title": "builder chart data read gap", "description": "The current builder capability cannot satisfy the requested workflow.", "tool_name": "chart_get", }, }, "workspace_icon_catalog_get": { "allowed_keys": [], "aliases": {}, "allowed_values": { "icon": list(WORKSPACE_ICON_NAMES), "color": list(WORKSPACE_ICON_COLORS), "generic_icon_names": list(GENERIC_WORKSPACE_ICON_NAMES), }, "execution_notes": [ "read this before creating app packages, apps, or portals when choosing supported workspace icons", "the CLI validates icon/color candidates but does not infer business defaults from resource names", "new app/package/portal creation requires explicit non-template icon + color", ], "minimal_example": { "profile": "default", }, }, "package_list": { "allowed_keys": ["trial_status", "query"], "aliases": {"trialStatus": "trial_status", "keyword": "query"}, "allowed_values": {"trial_status": ["all"]}, "execution_notes": [ "lists app packages visible to the current builder profile by calling backend GET /tag?trialStatus=...", "query is applied locally to package_id/tag_id/package_name/tag_name after /tag returns", "does not fall back to app list because app list cannot represent empty packages, duplicate package names, or package-level permissions", "returns package_id/package_name plus compatible tag_id/tag_name; use package_get for package detail before editing", "permission failures are returned as PACKAGE_LIST_FAILED with backend transport details", ], "minimal_example": { "profile": "default", "trial_status": "all", "query": "产品研发", }, }, "package_get": { "allowed_keys": ["package_id"], "aliases": {"packageId": "package_id"}, "allowed_values": {}, "execution_notes": [ "returns complete public package schema with package_id, package_name, icon, visibility, and items", "package_id maps internally to backend tagId; do not use tag_id in public calls", "visibility is normalized from backend MemberAuthInfoVO auth", ], "minimal_example": { "profile": "default", "package_id": 1001, }, }, "package_apply": { "allowed_keys": [ "package_id", "package_name", "icon", "color", "icon_name", "icon_color", "icon_config", "visibility", "items", "allow_detach", ], "aliases": { "packageId": "package_id", "packageName": "package_name", "iconName": "icon", "iconColor": "color", "icon_config.name": "icon", "icon_config.icon_name": "icon", "icon_config.color": "color", "icon_config.icon_color": "color", "allowDetach": "allow_detach", }, "allowed_values": { **deepcopy(_VISIBILITY_ALLOWED_VALUES), "workspace_icon.icon_names": list(WORKSPACE_ICON_NAMES), "workspace_icon.icon_colors": list(WORKSPACE_ICON_COLORS), "workspace_icon.generic_icon_names": list(GENERIC_WORKSPACE_ICON_NAMES), }, "execution_notes": [ "create or update package metadata, visibility, grouping, and ordering in one call", "create mode: package_name without package_id creates a package", "creating a package requires explicit icon + color; icon=template is blocked because it is too generic", "agent-facing primary icon shape is icon + color; icon_name/icon_color, icon_config, and icon={name,color} are compatibility aliases that normalize to icon/color", "updating a package preserves existing icon/color when omitted; explicit icon/color values are still validated", "call workspace_icon_catalog_get or `qingflow --json builder icon catalog` for supported icon/color candidates", "metadata keys omitted on update are preserved", "package_id maps internally to backend tagId; do not use tag_id in public calls", "items is a full package layout tree; omitting existing app/portal items is blocked unless allow_detach=true", "item shapes: {type:'app', app_key}, {type:'portal', dash_key}, or {type:'group', group_id?, name, items:[...]}", "allow_detach=true with flat app/portal items and no groups is treated as an explicit full replacement and skips current layout read plus final package readback", "layout apply calls backend package ordering (MoveGroupAuth), so it requires package edit_app permission even when items=[] only clears/deletes existing groups", *_VISIBILITY_EXECUTION_NOTES, ], "minimal_example": { "profile": "default", "package_id": 1001, "package_name": "项目管理", "visibility": deepcopy(_VISIBILITY_WORKSPACE_EXAMPLE), "items": [{"type": "app", "app_key": "APP_KEY", "title": "项目台账"}], "allow_detach": False, }, "specific_visibility_example": { "profile": "default", "package_id": 1001, "visibility": deepcopy(_VISIBILITY_SPECIFIC_EXAMPLE), }, "create_example": { "profile": "default", "package_name": "项目管理", "icon": "briefcase", "color": "azure", "visibility": deepcopy(_VISIBILITY_WORKSPACE_EXAMPLE), }, }, "solution_install": { "allowed_keys": ["solution_key", "being_copy_data", "solution_source"], "aliases": { "solutionKey": "solution_key", "beingCopyData": "being_copy_data", "copy_data": "being_copy_data", "copyData": "being_copy_data", "solutionSource": "solution_source", "source": "solution_source", }, "allowed_values": { "being_copy_data": [True, False], "solution_source": ["solutionDetail"], }, "execution_notes": [ "solution_install calls the backend app creation route with a solutionKey payload", "this is a write operation that may create multiple apps at once", "app_keys are verified from the backend response when available", ], "minimal_example": { "profile": "default", "solution_key": "cfqrhth99401", "being_copy_data": True, "solution_source": "solutionDetail", }, }, "member_search": { "allowed_keys": ["query", "page_num", "page_size", "contain_disable"], "aliases": {}, "allowed_values": {}, "minimal_example": { "profile": "default", "query": "严琪东", "page_num": 1, "page_size": 20, "contain_disable": False, }, }, "role_search": { "allowed_keys": ["keyword", "page_num", "page_size"], "aliases": {}, "allowed_values": {}, "minimal_example": { "profile": "default", "keyword": "项目经理", "page_num": 1, "page_size": 20, }, }, "role_create": { "allowed_keys": ["role_name", "member_uids", "member_emails", "member_names", "role_icon"], "aliases": {}, "allowed_values": {}, "minimal_example": { "profile": "default", "role_name": "研发负责人", "member_names": ["严琪东"], "member_uids": [], "member_emails": [], "role_icon": "ex-user-outlined", }, }, "app_resolve": { "allowed_keys": ["app_key", "app_name", "package_id"], "aliases": {"packageId": "package_id"}, "allowed_values": {}, "execution_notes": [ "use exactly one selector mode", "mode 1: app_key", "mode 2: app_name + package_id", ], "minimal_example": { "profile": "default", "app_key": "APP_KEY", }, "package_scoped_example": { "profile": "default", "app_name": "研发项目管理", "package_id": 1001, }, }, "app_release_edit_lock_if_mine": { "allowed_keys": ["app_key", "lock_owner_email", "lock_owner_name"], "aliases": {}, "allowed_values": {}, "execution_notes": [ "use this only after a builder write fails with APP_EDIT_LOCKED and the lock belongs to the current user", "supply the lock owner details from the failed write result for a safe release attempt", "after a successful release, retry the blocked builder write or publish verification call", ], "minimal_example": { "profile": "default", "app_key": "APP_KEY", "lock_owner_email": "user@example.com", "lock_owner_name": "当前用户", }, }, "button_style_catalog_get": { "allowed_keys": [], "aliases": {}, "allowed_values": { "preset.key": [item["key"] for item in BUTTON_STYLE_PRESETS], "icon": list(BUTTON_ICONS), "background_color": list(BUTTON_BACKGROUND_COLORS), "text_color": list(BUTTON_TEXT_COLORS), }, "execution_notes": [ "use this read-only tool before button writes when an agent needs a supported icon or color choice", "current frontend only supports button icons and button colors from this catalog", "text/icon color is unified through text_color; there is no separate icon_color", ], "minimal_example": { "profile": "default", }, }, "app_custom_buttons_apply": { "allowed_keys": [ "app_key", "upsert_buttons", "patch_buttons", "remove_buttons", "view_configs", "upsert_buttons[].client_key", "upsert_buttons[].button_id", "upsert_buttons[].button_text", "upsert_buttons[].style_preset", "upsert_buttons[].background_color", "upsert_buttons[].text_color", "upsert_buttons[].button_icon", "upsert_buttons[].trigger_action", "upsert_buttons[].trigger_link_url", "upsert_buttons[].trigger_add_data_config", "upsert_buttons[].trigger_add_data_config.target_app_key", "upsert_buttons[].trigger_add_data_config.field_mappings", "upsert_buttons[].trigger_add_data_config.field_mappings[].source_field", "upsert_buttons[].trigger_add_data_config.field_mappings[].target_field", "upsert_buttons[].trigger_add_data_config.default_values", "patch_buttons[].button_id", "patch_buttons[].button_text", "patch_buttons[].set", "patch_buttons[].unset", "view_configs[].view_key", "view_configs[].mode", "view_configs[].buttons", "view_configs[].buttons[].button_ref", "view_configs[].buttons[].placement", "view_configs[].buttons[].primary", "view_configs[].buttons[].button_limit", "view_configs[].buttons[].button_formula", "view_configs[].buttons[].button_formula_type", "view_configs[].buttons[].print_tpls", "remove_buttons[].button_id", "remove_buttons[].button_text", ], "aliases": { "upsertButtons": "upsert_buttons", "patchButtons": "patch_buttons", "removeButtons": "remove_buttons", "viewConfigs": "view_configs", "clientKey": "client_key", "buttonId": "button_id", "buttonText": "button_text", "stylePreset": "style_preset", "backgroundColor": "background_color", "textColor": "text_color", "buttonIcon": "button_icon", "triggerAction": "trigger_action", "triggerLinkUrl": "trigger_link_url", "triggerAddDataConfig": "trigger_add_data_config", "targetAppKey": "trigger_add_data_config.target_app_key", "fieldMappings": "trigger_add_data_config.field_mappings", "defaultValues": "trigger_add_data_config.default_values", "mode": "view_configs[].mode", "applyMode": "view_configs[].mode", "buttonRef": "view_configs[].buttons[].button_ref", "beingMain": "view_configs[].buttons[].primary", "buttonLimit": "view_configs[].buttons[].button_limit", "visibleWhen": "view_configs[].buttons[].button_limit", "buttonFormula": "view_configs[].buttons[].button_formula", "buttonFormulaType": "view_configs[].buttons[].button_formula_type", "printTpls": "view_configs[].buttons[].print_tpls", "externalQrobotConfig": "external_qrobot_config", "customButtonExternalQRobotRelationVO": "external_qrobot_config", "triggerWingsConfig": "trigger_wings_config", }, "allowed_values": { "upsert_buttons[].trigger_action": [member.value for member in PublicButtonTriggerAction], "upsert_buttons[].style_preset": [item["key"] for item in BUTTON_STYLE_PRESETS], "upsert_buttons[].button_icon": list(BUTTON_ICONS), "upsert_buttons[].background_color": list(BUTTON_BACKGROUND_COLORS), "upsert_buttons[].text_color": list(BUTTON_TEXT_COLORS), "view_configs[].mode": ["merge", "replace"], "view_configs[].buttons[].placement": [member.value for member in PublicButtonPlacement], }, "execution_notes": [ "this is the default custom-button write path; old list/get/create/update/delete tools are hidden from the public agent surface", "use patch_buttons for partial parameter replacement on existing buttons; the tool reads current button detail, merges patch_buttons[].set/unset, then submits the backend full-save payload internally", "button_id targets an existing button; without button_id, button_text is used as an exact unique upsert key, otherwise a new button is created", "for addData buttons, use trigger_add_data_config.target_app_key plus field_mappings/default_values; field_mappings compiles to backend queRelation", "field_mappings.source_field accepts source schema fields and supported system fields: 数据ID/row_record_id/apply_id/_id maps to current record id (-17), 编号/record_number maps to visible record number (0)", "to fill a target relation field with the current source record, map source_field='数据ID' to the target relation field; default_values is for static constants, not dynamic current-record values", "do not write raw que_relation unless maintaining a legacy config; field_mappings/default_values and que_relation are mutually exclusive", "permission split follows backend routes: upsert_buttons/patch_buttons/remove_buttons require EditAppAuth; view_configs also requires ViewManagementAuth (beingViewManageStatus), which falls back to DataManageAuth when advanced app permissions are not enabled", "view_configs binds custom buttons into views in the same apply call; button_ref may be a same-call client_key, a button_id, or an exact unique existing button_text", "view_configs[].view_key is the raw builder view key from app_get.views[].view_key; do not pass record-data view_id values like custom:VIEW_KEY", "view_configs[].buttons is required in merge mode; omitting buttons is blocked to avoid no-op writes and accidental publish", "view_configs[].mode defaults to merge; use mode=replace or an explicit empty buttons list to replace/clear a view's custom button bindings", "advanced view button binding supports button_limit, button_formula, button_formula_type, and print_tpls when visibility or print-template behavior is required", "default placements are header and detail; header maps to frontend top buttons", "placement=list configures backend INSIDE row/list buttons; header maps to TOP and detail maps to DETAIL", "remove_buttons supports button_id or exact unique button_text", "after a remove_buttons DELETE is sent, the tool verifies deletion by single button_id readback; removed[] returns delete_executed, readback_status, and safe_to_retry_delete=false", "if a removed button returns readback_status=unavailable or still_exists, treat the result as readback pending and do not blindly repeat the delete", "pure create/update with returned button ids and no view_configs/remove skips the final custom-button inventory read; response details.final_readback_skipped=true", "all operations share one edit context and publish after at least one write succeeds; there is no draft-only mode for this tool", "background_color and text_color cannot both be white", ], "minimal_example": { "profile": "default", "app_key": "APP_KEY", "upsert_buttons": [ { "button_text": "同步客户", "style_preset": "primary_blue", "button_icon": "ex-switch", "trigger_action": "link", "trigger_link_url": "https://example.com", } ], "patch_buttons": [ { "button_text": "同步客户", "set": {"trigger_link_url": "https://example.com/new"}, } ], "remove_buttons": [{"button_text": "旧按钮"}], "view_configs": [], }, "add_data_example": { "profile": "default", "app_key": "APP_KEY", "upsert_buttons": [ { "client_key": "add_order", "button_text": "新增订单", "style_preset": "neutral_outline", "button_icon": "ex-plus-circle", "trigger_action": "addData", "trigger_add_data_config": { "target_app_key": "TARGET_APP", "field_mappings": [{"source_field": "客户名称", "target_field": "客户"}], "default_values": {"状态": "待提交"}, }, } ], "remove_buttons": [], "view_configs": [ { "view_key": "VIEW_KEY", "buttons": [{"button_ref": "add_order", "placement": "detail", "primary": True}], } ], }, "relation_current_record_example": { "profile": "default", "app_key": "EMPLOYEE_APP", "upsert_buttons": [ { "client_key": "add_worklog", "button_text": "快捷添加工时", "style_preset": "neutral_outline", "button_icon": "ex-plus-circle", "trigger_action": "addData", "trigger_add_data_config": { "target_app_key": "WORKLOG_APP", "field_mappings": [{"source_field": "数据ID", "target_field": "关联员工"}], "default_values": {"状态": "待提交"}, }, } ], "view_configs": [ { "view_key": "RAW_VIEW_KEY", "buttons": [{"button_ref": "add_worklog", "placement": "detail", "primary": True}], } ], }, }, "app_associated_resources_apply": { "allowed_keys": [ "app_key", "upsert_resources", "patch_resources", "remove_associated_item_ids", "reorder_associated_item_ids", "view_configs", "upsert_resources[].client_key", "upsert_resources[].associated_item_id", "upsert_resources[].graph_type", "upsert_resources[].target_app_key", "upsert_resources[].view_key", "upsert_resources[].chart_key", "upsert_resources[].report_source", "upsert_resources[].match_mappings", "upsert_resources[].match_mappings[].target_field", "upsert_resources[].match_mappings[].source_field", "upsert_resources[].match_mappings[].value", "upsert_resources[].match_mappings[].values", "upsert_resources[].match_mappings[].operator", "upsert_resources[].match_rules", "patch_resources[].associated_item_id", "patch_resources[].set", "patch_resources[].unset", "view_configs[].view_key", "view_configs[].visible", "view_configs[].limit_type", "view_configs[].associated_item_ids", "view_configs[].associated_item_refs", ], "aliases": { "upsertResources": "upsert_resources", "patchResources": "patch_resources", "resources": "upsert_resources", "removeAssociatedItemIds": "remove_associated_item_ids", "reorderAssociatedItemIds": "reorder_associated_item_ids", "viewConfigs": "view_configs", "associatedItemId": "associated_item_id", "graphType": "graph_type", "targetAppKey": "target_app_key", "chartKey": "chart_key", "chartId": "chart_key", "viewKey": "view_key", "viewgraphKey": "view_key", "reportSource": "report_source", "matchMappings": "match_mappings", "matchRules": "match_rules", "associatedItemRefs": "associated_item_refs", "associatedItemIds": "associated_item_ids", }, "allowed_values": { "upsert_resources[].graph_type": ["chart", "view"], "upsert_resources[].report_source": ["app", "dataset"], "upsert_resources[].match_mappings[].operator": [ "eq", "neq", "in", "contains", "gte", "lte", "is_empty", "not_empty", ], "view_configs[].limit_type": ["all", "select"], }, "execution_notes": [ "this tool manages Qingflow in-app associated report/view display; it does not create or edit QingBI report bodies/configs", "create or edit app-source BI report bodies first with app_charts_apply, then attach the resulting chart_id here with graph_type=chart; dataset BI reports can only be attached when they already exist", "this is the default associated report/view path; it manages both the app-level associated resource pool and per-view display config", "permission split follows backend routes: upsert_resources/patch_resources/remove/reorder require EditAppAuth; view_configs require ViewManagementAuth (beingViewManageStatus), which falls back to DataManageAuth when advanced app permissions are not enabled", "use patch_resources for partial parameter replacement on existing associated resources; the tool reads the current resource including backend-required raw fields, merges patch_resources[].set/unset, then submits the backend full-save payload internally", "associated_item_id is form_asos_chart.id from app_get.associated_resources[].associated_item_id; view_configs/remove/reorder may also pass an existing associated resource's chart_id/chart_key/view_key and the tool resolves it to the internal id", "before creating an associated resource, read app_get.associated_resources and reuse an existing item with patch_resources when target_app_key + view_key/chart_key already matches; repeated upsert_resources without associated_item_id can create duplicate associated items", "graph_type=view uses view_key and internally compiles to the Qingflow view source; graph_type=chart uses chart_key and defaults to report_source=app", "report_source=app maps to BI_QINGFLOW; report_source=dataset maps to BI_DATASET for associating an existing dataset report; do not pass raw backend sourceType", "use match_mappings for associated view/report filtering; dynamic conditions use source_field and static conditions use value/values", "match_mappings.source_field accepts source schema fields plus system fields 数据ID(-17) and 编号(0); match_mappings compiles to backend matchRules", "match_mappings uses the same public operator semantics as view/chart filters: eq, neq, in, contains, gte, lte, is_empty, not_empty; field_name/field are aliases of target_field", "static multi-value conditions use values; is_empty/not_empty conditions do not need source_field/value/values", "do not write raw match_rules unless preserving a legacy backend config; match_mappings and match_rules are mutually exclusive", "client_key only lets a view_config reference a resource created earlier in the same apply call through associated_item_refs; it is not persisted and cannot deduplicate later apply calls", "remove_associated_item_ids sends DELETE and verifies deletion with one associated-resource pool readback because the backend has no confirmed single-item GET; removed[] returns delete_executed, readback_status, and safe_to_retry_delete=false", "if an associated resource delete returns readback_status=unavailable or still_exists, treat the result as readback pending and do not blindly repeat the delete", "pure create/update with returned associated_item_id values and no view_configs/remove/reorder skips the final associated-resource pool read; response details.final_readback_skipped=true", "this tool publishes after at least one write succeeds; there is no draft-only mode", "visible=false hides the associated-resource area without clearing previous selected ids; visible=true with limit_type=all shows the whole app-level pool", ], "minimal_example": { "profile": "default", "app_key": "APP_KEY", "upsert_resources": [ { "client_key": "customer_view", "graph_type": "view", "target_app_key": "TARGET_APP", "view_key": "VIEW_KEY", "match_mappings": [{"target_field": "关联员工", "source_field": "数据ID"}], } ], "patch_resources": [ { "associated_item_id": 123, "set": {"match_mappings": [{"target_field": "状态", "value": "待提交"}]}, } ], "remove_associated_item_ids": [], "reorder_associated_item_ids": [], "view_configs": [ { "view_key": "MAIN_VIEW", "visible": True, "limit_type": "select", "associated_item_refs": ["customer_view"], } ], }, }, "app_schema_plan": { "allowed_keys": ["app_key", "package_id", "app_name", "icon", "color", "visibility", "add_fields", "update_fields", "remove_fields"], "aliases": { "app_title": "app_name", "title": "app_name", "packageId": "package_id", "field.title": "field.name", "field.label": "field.name", "field.fields": "field.subfields", "field.type_id": "field.type", "field.relationMode": "field.relation_mode", "field.selection_mode": "field.relation_mode", "field.selectionMode": "field.relation_mode", "field.multiple": "field.relation_mode", "field.allow_multiple": "field.relation_mode", "field.optional_data_num": "field.relation_mode", "field.optionalDataNum": "field.relation_mode", "field.departmentScope": "field.department_scope", "field.remoteLookupConfig": "field.remote_lookup_config", "field.qLinkerBinding": "field.q_linker_binding", "field.codeBlockConfig": "field.code_block_config", "field.codeBlockBinding": "field.code_block_binding", "field.autoTrigger": "field.auto_trigger", "field.customBtnTextStatus": "field.custom_button_text_enabled", "field.customBtnText": "field.custom_button_text", "field.subfieldUpdates": "field.subfield_updates", "field.data_title": "field.as_data_title", "field.dataTitle": "field.as_data_title", "field.data_cover": "field.as_data_cover", "field.dataCover": "field.as_data_cover", "field.asDataTitle": "field.as_data_title", "field.asDataCover": "field.as_data_cover", }, "allowed_values": { "field.type": [member.value for member in PublicFieldType], "field.relation_mode": [member.value for member in PublicRelationMode], "field_type_ids": sorted(FIELD_TYPE_ID_ALIASES.keys()), "workspace_icon.icon_names": list(WORKSPACE_ICON_NAMES), "workspace_icon.icon_colors": list(WORKSPACE_ICON_COLORS), "workspace_icon.generic_icon_names": list(GENERIC_WORKSPACE_ICON_NAMES), **deepcopy(_VISIBILITY_ALLOWED_VALUES), }, "execution_notes": [ "create mode may set visibility for the new app; edit mode may update visibility on an existing app", "create mode should include explicit non-template icon + color; apply mode enforces this before writing", "call workspace_icon_catalog_get or `qingflow --json builder icon catalog` for supported icon/color candidates", "do not add form fields named 数据ID, 编号, 申请人, 申请时间, 创建人, 创建时间, 提交人, 提交时间, 更新时间, 更新人, 当前流程状态, 当前处理人, 当前处理节点, or 流程标题; these are Qingflow built-in system fields, not fields to create", *_VISIBILITY_EXECUTION_NOTES, ], "minimal_example": { "profile": "default", "app_name": "研发项目管理", "package_id": 1001, "icon": "briefcase", "color": "emerald", "visibility": deepcopy(_VISIBILITY_WORKSPACE_EXAMPLE), "add_fields": [{"name": "项目名称", "type": "text", "data_title": True}], "update_fields": [], "remove_fields": [], }, "relation_example": { "profile": "default", "app_key": "APP_ITERATION", "add_fields": [ { "name": "需求反馈引用", "type": "relation", "target_app_key": "APP_FEEDBACK", "relation_mode": "multiple", "display_field": {"name": "反馈标题"}, "visible_fields": [{"name": "反馈标题"}, {"name": "优先级判断"}], } ], "update_fields": [], "remove_fields": [], }, }, "app_schema_apply": { "allowed_keys": [ "app_key", "package_id", "app_name", "app_title", "icon", "color", "icon_name", "icon_color", "icon_config", "visibility", "publish", "form", "add_fields", "update_fields", "remove_fields", "apps", "apps[].client_key", "apps[].app_key", "apps[].app_name", "apps[].icon", "apps[].color", "apps[].icon_name", "apps[].icon_color", "apps[].icon_config", "apps[].visibility", "apps[].form", "apps[].form[].section", "apps[].form[].rows", "apps[].add_fields", "apps[].update_fields", "apps[].remove_fields", "apps[].add_fields[].target_app_ref", "apps[].add_fields[].target_app", ], "aliases": { "app_title": "app_name", "title": "app_name", "packageId": "package_id", "apps[].clientKey": "apps[].client_key", "apps[].appKey": "apps[].app_key", "apps[].appName": "apps[].app_name", "apps[].appTitle": "apps[].app_name", "apps[].iconName": "apps[].icon", "apps[].iconColor": "apps[].color", "apps[].icon_config.name": "apps[].icon", "apps[].icon_config.color": "apps[].color", "field.targetAppRef": "field.target_app_ref", "field.targetAppClientKey": "field.target_app_ref", "field.targetApp": "field.target_app", "field.options[].label": "field.options[]", "field.options[].value": "field.options[]", "field.options[].optValue": "field.options[]", "field.title": "field.name", "field.label": "field.name", "field.fields": "field.subfields", "field.type_id": "field.type", "field.relationMode": "field.relation_mode", "field.selection_mode": "field.relation_mode", "field.selectionMode": "field.relation_mode", "field.multiple": "field.relation_mode", "field.allow_multiple": "field.relation_mode", "field.optional_data_num": "field.relation_mode", "field.optionalDataNum": "field.relation_mode", "field.departmentScope": "field.department_scope", "field.remoteLookupConfig": "field.remote_lookup_config", "field.qLinkerBinding": "field.q_linker_binding", "field.codeBlockConfig": "field.code_block_config", "field.codeBlockBinding": "field.code_block_binding", "field.autoTrigger": "field.auto_trigger", "field.customBtnTextStatus": "field.custom_button_text_enabled", "field.customBtnText": "field.custom_button_text", "field.subfieldUpdates": "field.subfield_updates", "field.data_title": "field.as_data_title", "field.dataTitle": "field.as_data_title", "field.data_cover": "field.as_data_cover", "field.dataCover": "field.as_data_cover", "field.asDataTitle": "field.as_data_title", "field.asDataCover": "field.as_data_cover", }, "allowed_values": { "field.type": [member.value for member in PublicFieldType], "field.type_aliases": {alias: field_type.value for alias, field_type in sorted(FIELD_TYPE_ALIASES.items())}, "field.relation_mode": [member.value for member in PublicRelationMode], "field.department_scope.mode": ["all", "custom"], "field.code_block_binding.outputs.target_field.type": list(INTEGRATION_OUTPUT_TARGET_FIELD_TYPES), "field.q_linker_binding.outputs.target_field.type": list(INTEGRATION_OUTPUT_TARGET_FIELD_TYPES), "field_type_ids": sorted(FIELD_TYPE_ID_ALIASES.keys()), **deepcopy(_VISIBILITY_ALLOWED_VALUES), }, "execution_notes": [ "use exactly one resource mode", "edit mode: app_key, optional app_name to rename the existing app", "create mode: package_id + app_name", "create mode submits a create request directly after basic package add_app permission check; it does not scan app_name duplicates or reuse an existing app by name", "create mode follows backend CreateAppBean: package add_app permission is checked on the target package; package edit_app is not required for the create precheck", "multi-app mode: pass package_id + apps[]; items with app_name and no app_key are created, items with app_key update existing apps", "CLI --apps-file primary shape is {package_id, apps:[...]}; raw app arrays and singleton wrapper arrays are compatibility paths, not recommended examples", "create app schemas with form: form[].section names form sections; form[].rows[][] contains field objects; builder splits form into field creation plus form layout apply", "single-app CLI primary shape is --form-file; multi-app primary shape is apps[].form inside --apps-file", "multi-app mode validates static payload errors before writing: duplicate client_key/app_name, missing data title on new apps, and unresolved target_app_ref/target_app", "multi-app relation fields may use target_app_ref to point at another apps[].client_key; the tool creates/resolves app shells first and compiles it to target_app_key", "multi-app relation fields may also use target_app with another apps[].app_name; prefer target_app_ref/client_key when names may collide", "multi-app schema writes run in dependency waves: apps without same-batch relation dependencies write first, then dependent apps write in parallel by layer; response.parallelism includes schema_wave_count/schema_wave_sizes", "multi-app mode is not transactional; read created_app_keys and apps[].status before retrying, and retry only failed app items", "create mode defaults new app visibility to workspace/not when visibility is omitted; edit mode preserves current visibility when omitted", "create mode requires explicit icon + color; icon=template is blocked because it is too generic", "agent-facing primary icon shape is icon + color; icon_name/icon_color, icon_config, and icon={name,color} are compatibility aliases that normalize to icon/color", "multi-app create mode requires each new app item to include a distinct non-template icon and a valid color", "agent-friendly field type aliases are normalized before writing: multiline/multiline_text/textarea -> long_text; select/single_choice/dropdown -> single_select; multi_choice/multiple_choice/checkbox -> multi_select", "single_select and multi_select options accept strings or objects such as {label,value}; builder normalizes them to option labels before writing", "edit mode preserves existing icon/color when omitted; explicit icon/color values are still validated", "call workspace_icon_catalog_get or `qingflow --json builder icon catalog` for supported icon/color candidates", "do not add form fields named 数据ID, 编号, 申请人, 申请时间, 创建人, 创建时间, 提交人, 提交时间, 更新时间, 更新人, 当前流程状态, 当前处理人, 当前处理节点, or 流程标题; these are Qingflow built-in system fields, not fields to create", *_VISIBILITY_EXECUTION_NOTES, "update_fields is the field-level partial update path; it reads current form schema and preserves untouched field config", "an app may contain multiple relation fields; do not downgrade secondary relations to text just because more than one relation exists", "backend relation errors are returned as normal schema apply failures with backend_code/details; they are not normalized into a one-relation modeling fallback", "relation_mode=multiple maps to referenceConfig.optionalDataNum=0", "relation fields now require both display_field and visible_fields in MCP/CLI payloads", "if relation target metadata lookup is blocked by 40161/40002/40027, explicit display_field.name and visible_fields[].name let builder degrade verification and still continue schema write", "relation write readback returns details.relation_readback_matrix; mismatches include details.relation_repair_plan with a minimal update_fields relation patch and data_impact", "update_fields[].set.subfield_updates is the safe patch path for editing existing subtable child fields without rebuilding the entire subtable", "subfield_updates only supports safe child overlays: name, required, description, and nested subfield_updates", "set.subfields remains the full replace/rebuild path for a subtable and is higher risk when hidden relation/reference children exist", "department fields accept department_scope with mode=all or mode=custom; custom scope requires explicit departments[].dept_id and optional include_sub_departs", "q_linker_binding lets you declare request config, dynamic inputs, alias parsing, and target-field bindings in one step; builder writes remoteLookupConfig plus the existing backend relation-default and questionRelations structures", "code_block_binding lets you declare inputs, code, alias parsing, and target-field bindings in one step; builder writes codeBlockConfig plus the existing backend relation-default and questionRelations structures", "builder configures code blocks only; it does not execute or trigger code blocks", "code block outputs must be emitted through qf_output assignment; use qf_output = {...} or assign qf_output after building the result object, never const/let qf_output =", "builder automatically normalizes const/let qf_output assignments on write and rejects output-bound code blocks that still do not contain a valid qf_output assignment", "code_block_binding and q_linker_binding target fields are limited to text, long_text, number, amount, date, datetime, single_select, multi_select, and boolean", "data title is required: mark exactly one top-level field with as_data_title=true; use a readable name/number/date-like field", "data cover is optional: mark at most one top-level attachment field with as_data_cover=true", ], "minimal_example": { "profile": "default", "app_name": "研发项目管理", "package_id": 1001, "icon": "briefcase", "color": "emerald", "visibility": deepcopy(_VISIBILITY_WORKSPACE_EXAMPLE), "publish": True, "form": [ { "section": "基础信息", "rows": [ [ {"name": "项目名称", "type": "text", "data_title": True}, {"name": "项目封面", "type": "attachment", "data_cover": True}, ] ], } ], "update_fields": [], "remove_fields": [], }, "multi_app_example": { "profile": "default", "package_id": 1001, "publish": True, "apps": [ { "client_key": "employee", "app_name": "员工花名册", "icon": "business-personalcard", "color": "emerald", "form": [ { "section": "基础信息", "rows": [ [ {"name": "员工名称", "type": "text", "data_title": True}, {"name": "员工照片", "type": "attachment", "data_cover": True}, ] ], } ], }, { "client_key": "worklog", "app_name": "工时表", "icon": "clock", "color": "blue", "form": [ { "section": "基础信息", "rows": [ [{"name": "工时标题", "type": "text", "data_title": True}], [ { "name": "关联员工", "type": "relation", "target_app_ref": "employee", "display_field": {"name": "员工名称"}, "visible_fields": [{"name": "员工名称"}], } ], ], } ], }, ], }, "rename_example": { "profile": "default", "app_key": "APP_PROJECT", "app_name": "项目主数据", "publish": True, "add_fields": [], "update_fields": [], "remove_fields": [], }, "relation_example": { "profile": "default", "app_key": "APP_ITERATION", "publish": True, "add_fields": [ { "name": "需求反馈引用", "type": "relation", "target_app_key": "APP_FEEDBACK", "relation_mode": "multiple", "display_field": {"name": "反馈标题"}, "visible_fields": [{"name": "反馈标题"}, {"name": "优先级判断"}], } ], "update_fields": [], "remove_fields": [], }, "subfield_update_example": { "profile": "default", "app_key": "APP_REWARD", "publish": True, "add_fields": [], "update_fields": [ { "selector": {"que_id": 441305044}, "set": { "subfield_updates": [ { "selector": {"que_id": 441305045}, "set": {"name": "景品名"}, } ] }, } ], "remove_fields": [], }, "department_scope_example": { "profile": "default", "app_key": "APP_LEAD", "publish": True, "add_fields": [ { "name": "归属部门", "type": "department", "department_scope": { "mode": "custom", "departments": [{"dept_id": 334952, "dept_name": "生产"}], "include_sub_departs": False, }, } ], "update_fields": [], "remove_fields": [], }, "code_block_example": { "profile": "default", "app_key": "APP_SCRIPT", "publish": True, "add_fields": [ {"name": "客户等级", "type": "text"}, { "name": "查询代码块", "type": "code_block", "code_block_binding": { "inputs": [ {"field": {"name": "客户名称"}, "var": "customerName"}, {"field": {"name": "预算金额"}, "var": "budget"}, ], "code": "qf_output = {}; qf_output.customerLevel = budget > 100000 ? 'A' : 'B';", "auto_trigger": True, "custom_button_text_enabled": True, "custom_button_text": "评估客户", "outputs": [ {"alias": "customerLevel", "path": "$.customerLevel", "target_field": {"name": "客户等级"}}, ], }, } ], "update_fields": [], "remove_fields": [], }, "q_linker_example": { "profile": "default", "app_key": "APP_CUSTOMER", "publish": True, "add_fields": [ {"name": "客户名称", "type": "text"}, {"name": "企业名称", "type": "text"}, {"name": "统一社会信用代码", "type": "text"}, { "name": "企业信息查询", "type": "q_linker", "q_linker_binding": { "inputs": [ {"field": {"name": "客户名称"}, "key": "keyword", "source": "query_param"}, ], "request": { "url": "https://example.com/company/search", "method": "GET", "headers": [], "query_params": [], "body_type": 1, "url_encoded_value": [], "result_type": 1, "auto_trigger": True, "custom_button_text_enabled": True, "custom_button_text": "查询企业信息", }, "outputs": [ {"alias": "company_name", "path": "$.data.name", "target_field": {"name": "企业名称"}}, {"alias": "credit_code", "path": "$.data.creditCode", "target_field": {"name": "统一社会信用代码"}}, ], }, }, ], "update_fields": [], "remove_fields": [], }, }, "app_layout_plan": { "allowed_keys": ["app_key", "mode", "sections", "preset"], "aliases": {"overwrite": "replace", "sectionId": "section_id"}, "section_allowed_keys": ["type", "paragraph_id", "section_id", "title", "rows"], "section_aliases": { "name": "title", "paragraphId": "paragraph_id", "sectionId": "section_id", "fields": "rows", "field_ids": "rows", "columns": "rows_chunk_size", }, "allowed_values": {"mode": [member.value for member in LayoutApplyMode], "preset": [member.value for member in LayoutPreset]}, "minimal_section_example": {"type": "paragraph", "paragraph_id": "basic", "title": "基础信息", "rows": [["字段A", "字段B", "字段C", "字段D"]]}, "minimal_example": { "profile": "default", "app_key": "APP_KEY", "mode": "merge", "sections": [{"type": "paragraph", "paragraph_id": "basic", "title": "基础信息", "rows": [["字段A", "字段B", "字段C", "字段D"]]}], }, "preset_example": {"profile": "default", "app_key": "APP_KEY", "mode": "merge", "preset": "balanced", "sections": []}, }, "app_layout_apply": { "allowed_keys": ["app_key", "mode", "publish", "sections"], "aliases": {"overwrite": "replace", "sectionId": "section_id"}, "section_allowed_keys": ["type", "paragraph_id", "section_id", "title", "rows"], "section_aliases": { "name": "title", "paragraphId": "paragraph_id", "sectionId": "section_id", "fields": "rows", "field_ids": "rows", "columns": "rows_chunk_size", }, "allowed_values": {"mode": [member.value for member in LayoutApplyMode]}, "execution_notes": [ "mode=merge is the layout partial update path: mentioned sections are merged and unmentioned fields are preserved", "mode=replace is full layout replacement and should be used only when intentionally rewriting all sections", "layout verification is split into layout_verified and layout_summary_verified", "LAYOUT_SUMMARY_UNVERIFIED means raw form readback is stronger than the compact summary", ], "minimal_section_example": {"type": "paragraph", "paragraph_id": "basic", "title": "基础信息", "rows": [["字段A", "字段B", "字段C", "字段D"]]}, "minimal_example": { "profile": "default", "app_key": "APP_KEY", "mode": "merge", "publish": True, "sections": [{"type": "paragraph", "paragraph_id": "basic", "title": "基础信息", "rows": [["项目名称", "项目负责人", "项目阶段", "优先级"]]}], }, }, "app_flow_plan": { "allowed_keys": ["app_key", "mode", "nodes", "transitions", "preset"], "aliases": { "overwrite": "replace", "base_preset": "preset", "default_approval": "basic_approval", "node.role_names": "node.assignees.role_names", "node.role_ids": "node.assignees.role_ids", "node.member_names": "node.assignees.member_names", "node.member_emails": "node.assignees.member_emails", "node.member_uids": "node.assignees.member_uids", "node.editable_fields": "node.permissions.editable_fields", "default_approval": "basic_approval", }, "allowed_values": { "mode": ["replace"], "preset": [member.value for member in FlowPreset], "node.type": PUBLIC_STABLE_FLOW_NODE_TYPES, }, "dependency_hints": [ "approval-style workflows require an explicit business status select field before app_flow_apply, such as 状态 / 处理状态 / 审批状态 / 工单状态", "approve/fill/copy nodes require at least one assignee", ], "execution_notes": [ "app_flow_plan is private compatibility plumbing, not an agent-facing workflow path; use app_flow_get/app_flow_get_schema plus app_flow_apply spec or patch_nodes", "do not create platform system workflow fields such as 当前流程状态 / 当前处理人 / 当前处理节点 / 流程标题; use an explicit business status field instead", ], "minimal_example": { "profile": "default", "app_key": "APP_KEY", "mode": "replace", "preset": "basic_approval", "nodes": [ { "id": "approve_1", "type": "approve", "name": "部门审批", "assignees": {"role_names": ["项目经理"]}, "permissions": {"editable_fields": ["状态", "审批意见"]}, } ], "transitions": [], }, }, "app_flow_apply": { "allowed_keys": [ "app_key", "publish", "spec", "patch_nodes", "idempotency_key", "schema_version", ], "aliases": { "patchNodes": "patch_nodes", "idempotencyKey": "idempotency_key", "schemaVersion": "schema_version", }, "allowed_values": {}, "dependency_hints": [ "approval-style workflows require an explicit business status select field before app_flow_apply, such as 状态 / 处理状态 / 审批状态 / 工单状态", "WorkflowSpec nodes should declare sync=true for routing nodes and explicit assignees for approval/filling/cc nodes", ], "execution_notes": [ "preferred full workflow path is app_flow_get_schema/app_flow_get -> app_flow_apply with spec", "use patch_nodes for targeted existing-node edits; each item uses id plus set/unset", "public workflow writes use only spec or patch_nodes", "do not mix spec and patch_nodes in one call", "do not create platform system workflow fields such as 当前流程状态 / 当前处理人 / 当前处理节点 / 流程标题; use an explicit business status field instead", "workflow readback is app_flow_get; use app_flow_get_schema when raw spec shape is unclear", ], "minimal_example": { "profile": "default", "app_key": "APP_KEY", "publish": True, "spec": {"schemaVersion": "1", "nodes": [], "edges": []}, }, "patch_nodes_example": { "profile": "default", "app_key": "APP_KEY", "publish": True, "patch_nodes": [{"id": "approve_1", "set": {"name": "主管审批"}}], }, }, "app_views_plan": { "allowed_keys": [ "app_key", "upsert_views", "patch_views", "remove_views", "preset", "upsert_views[].view_key", "upsert_views[].name", "upsert_views[].type", "upsert_views[].columns", "upsert_views[].filters", "upsert_views[].buttons", "upsert_views[].action_buttons", "upsert_views[].action_buttons_mode", "upsert_views[].visibility", "upsert_views[].query_conditions", "patch_views[].view_key", "patch_views[].name", "patch_views[].set", "patch_views[].set.action_buttons", "patch_views[].set.action_buttons_mode", "patch_views[].unset", ], "aliases": { "fields": "columns", "column_names": "columns", "columnNames": "columns", "viewKey": "view_key", "tableView": "table", "cardView": "card", "kanban": "board", "filter_rules": "filters", "filterRules": "filters", "queryConditions": "query_conditions", "query_condition": "query_conditions", "queryCondition": "query_conditions", "patchViews": "patch_views", "startField": "start_field", "endField": "end_field", "titleField": "title_field", "buttons[].buttonType": "buttons[].button_type", "buttons[].configType": "buttons[].config_type", "buttons[].buttonId": "buttons[].button_id", "buttons[].beingMain": "buttons[].being_main", "buttons[].buttonLimit": "buttons[].button_limit", "buttons[].buttonFormula": "buttons[].button_formula", "buttons[].buttonFormulaType": "buttons[].button_formula_type", "buttons[].printTpls": "buttons[].print_tpls", "actionButtons": "action_buttons", "actionButtonsMode": "action_buttons_mode", "action_buttons[].button_text": "action_buttons[].text", "action_buttons[].trigger_action": "action_buttons[].action", "action_buttons[].trigger_link_url": "action_buttons[].url", "action_buttons[].visibleWhen": "action_buttons[].visible_when", "action_buttons[].fieldMappings": "action_buttons[].field_mappings", "action_buttons[].defaultValues": "action_buttons[].default_values", }, "allowed_values": { "preset": [member.value for member in ViewsPreset], "view.type": [member.value for member in PublicViewType], "view.filter.operator": [member.value for member in ViewFilterOperator], "view.buttons.button_type": ["SYSTEM", "CUSTOM"], "view.buttons.config_type": ["TOP", "DETAIL", "INSIDE"], "view.action_buttons.action": ["add_data", "link", "qRobot", "wings"], "view.action_buttons.placement": [member.value for member in PublicButtonPlacement], "view.action_buttons_mode": ["merge", "replace"], **deepcopy(_VISIBILITY_ALLOWED_VALUES), }, "execution_notes": [ "creating a new view follows backend createViewgraphConfig and requires both ViewManagementAuth and DataManageAuth; updating/deleting existing views only requires ViewManagementAuth", "upsert_views[].visibility may set per-view visibility; omit it to preserve an existing view's auth or default a new view to workspace/not", "filters are saved fixed filters that apply when the view opens; query_conditions configure the frontend query panel and only apply after a user enters query values", "upsert_views[].query_conditions.rows is a layout matrix of query-panel supported field names; it is compiled to backend queryCondition queIds", "do not put relation/attachment/subtable/code_block/q_linker/address fields in query_conditions; use filters for fixed filters or app_associated_resources_apply.match_mappings for current-record relation/report matching", "use patch_views for partial parameter replacement on existing views; the tool reads current config, merges patch_views[].set/unset, then submits the backend full-save payload internally", "business buttons should be declared with action_buttons on the target view; app_views_apply writes the view first, then compiles action_buttons through the custom button writer and binds them to the resulting view_key", "action_buttons_mode defaults to merge; replace rewrites this view's custom button bindings, and replace plus action_buttons=[] clears this view's custom button bindings without deleting button bodies", "legacy upsert_views[].buttons is a compatibility path for raw full replacement of existing view button config; do not use it as the normal agent path", "remove_views accepts a raw view_key or an exact unique view name; after DELETE the tool verifies deletion by single view_key readback, not by a full app view list", "deleted views return verification.by_view[].delete_executed, readback_status, and safe_to_retry_delete=false; if readback is pending, do not blindly repeat the delete", "do not create business views named 全部数据, 我的数据, 我发起的, 待办, 已办, or 抄送; these are built-in system/default views. Use business-specific names for new views, and pass the existing raw view_key or patch_views when changing a built-in view", "new views created by app_views_apply default associated report/view display to visible with limit_type=all; existing views preserve their current associated display unless patch_views/upsert_views explicitly changes associated_resources", "associated report/view resource pool and per-view selected resources are configured through app_associated_resources_apply; app_views_apply only keeps legacy associated_resources input compatible", "for multi-value operators such as in, pass values as a list; value may also be used as an alias when it already contains a list", *_VISIBILITY_EXECUTION_NOTES, ], "minimal_example": { "profile": "default", "app_key": "APP_KEY", "upsert_views": [{"name": "项目台账视图", "type": "table", "columns": ["项目名称"], "visibility": deepcopy(_VISIBILITY_WORKSPACE_EXAMPLE)}], "remove_views": [], }, "query_conditions_example": { "profile": "default", "app_key": "APP_KEY", "patch_views": [ { "view_key": "VIEW_KEY", "set": { "query_conditions": { "enabled": True, "rows": [["客户名称", "负责人"], ["创建时间"]], } }, } ], "remove_views": [], }, "action_buttons_example": { "profile": "default", "app_key": "APP_KEY", "upsert_views": [ { "name": "生产工单执行视图", "type": "table", "columns": ["工单编号", "产品", "状态", "负责人"], "action_buttons": [ { "text": "开始生产", "action": "link", "url": "https://example.com/start", "placement": "list", "visible_when": [{"field_name": "状态", "operator": "eq", "value": "待排产"}], } ], } ], "remove_views": [], }, "full_upsert_query_conditions_example": { "profile": "default", "app_key": "APP_KEY", "upsert_views": [ { "name": "客户查询视图", "type": "table", "columns": ["客户名称", "负责人", "客户状态", "创建时间"], "filters": [{"field_name": "客户状态", "operator": "eq", "value": "有效"}], "query_conditions": { "enabled": True, "exact": False, "hide_before_query": False, "rows": [["客户名称", "负责人"], ["创建时间"]], }, } ], "remove_views": [], }, "query_conditions_field_rules": { "supported_field_types": ["text", "long_text", "number", "amount", "date", "datetime", "single_select", "multi_select", "phone", "email", "boolean", "member", "department"], "unsupported_field_types": ["relation", "attachment", "subtable", "address", "code_block", "q_linker"], "use_instead": { "fixed_filter": "filters", "current_record_related_report_or_view": "app_associated_resources_apply.match_mappings", }, }, "gantt_example": { "profile": "default", "app_key": "APP_KEY", "upsert_views": [ { "name": "项目甘特图", "type": "gantt", "columns": ["项目名称", "开始日期", "结束日期"], "start_field": "开始日期", "end_field": "结束日期", "title_field": "项目名称", "filters": [{"field_name": "状态", "operator": "eq", "value": "进行中"}], } ], "remove_views": [], }, }, "app_views_apply": { "allowed_keys": [ "views", "views[].operation", "views[].app_key", "views[].view_key", "views[].name", "views[].type", "views[].columns", "views[].filters", "views[].action_buttons", "views[].set", "views[].unset", "app_key", "publish", "upsert_views", "patch_views", "remove_views", "upsert_views[].view_key", "upsert_views[].name", "upsert_views[].type", "upsert_views[].columns", "upsert_views[].filters", "upsert_views[].buttons", "upsert_views[].action_buttons", "upsert_views[].action_buttons_mode", "upsert_views[].visibility", "upsert_views[].query_conditions", "patch_views[].view_key", "patch_views[].name", "patch_views[].set", "patch_views[].set.action_buttons", "patch_views[].set.action_buttons_mode", "patch_views[].unset", ], "aliases": { "fields": "columns", "column_names": "columns", "columnNames": "columns", "viewKey": "view_key", "tableView": "table", "cardView": "card", "kanban": "board", "filter_rules": "filters", "filterRules": "filters", "queryConditions": "query_conditions", "query_condition": "query_conditions", "queryCondition": "query_conditions", "views[].op": "views[].operation", "views[].fields": "views[].columns", "views[].viewKey": "views[].view_key", "views[].actionButtons": "views[].action_buttons", "patchViews": "patch_views", "startField": "start_field", "endField": "end_field", "titleField": "title_field", "buttons[].buttonType": "buttons[].button_type", "buttons[].configType": "buttons[].config_type", "buttons[].buttonId": "buttons[].button_id", "buttons[].beingMain": "buttons[].being_main", "buttons[].buttonLimit": "buttons[].button_limit", "buttons[].buttonFormula": "buttons[].button_formula", "buttons[].buttonFormulaType": "buttons[].button_formula_type", "buttons[].printTpls": "buttons[].print_tpls", "actionButtons": "action_buttons", "actionButtonsMode": "action_buttons_mode", "action_buttons[].button_text": "action_buttons[].text", "action_buttons[].trigger_action": "action_buttons[].action", "action_buttons[].trigger_link_url": "action_buttons[].url", "action_buttons[].visibleWhen": "action_buttons[].visible_when", "action_buttons[].fieldMappings": "action_buttons[].field_mappings", "action_buttons[].defaultValues": "action_buttons[].default_values", }, "allowed_values": { "views[].operation": ["upsert", "patch", "remove"], "view.type": [member.value for member in PublicViewType], "view.filter.operator": [member.value for member in ViewFilterOperator], "view.buttons.button_type": ["SYSTEM", "CUSTOM"], "view.buttons.config_type": ["TOP", "DETAIL", "INSIDE"], "view.action_buttons.action": ["add_data", "link", "qRobot", "wings"], "view.action_buttons.placement": [member.value for member in PublicButtonPlacement], "view.action_buttons_mode": ["merge", "replace"], **deepcopy(_VISIBILITY_ALLOWED_VALUES), }, "execution_notes": [ "preferred batch input is views[]; each item has app_key and operation=upsert/patch/remove, and the same batch may mix same-app and cross-app views", f"views[] jobs run concurrently with an internal limit of {VIEW_APPLY_PARALLEL_LIMIT}; results are returned in input order", "apply may return partial_success when some views land and others fail", "creating a new view follows backend createViewgraphConfig and requires both ViewManagementAuth and DataManageAuth; updating/deleting existing views only requires ViewManagementAuth", "when duplicate view names exist, supply view_key to target the exact view", "read back app_get after any failed or partial view apply", "view existence verification and saved-filter verification are separate; treat filters as unverified until verification.view_filters_verified is true", "buttons omitted preserves existing button config; buttons=[] clears all buttons; buttons=[...] replaces the full button config", "business buttons should be declared with action_buttons on the target view; app_views_apply writes the view first, then compiles action_buttons through the custom button writer and binds them to the resulting view_key", "patch_views[].set.action_buttons can add buttons to an existing view without rewriting columns/filters; action_buttons_mode defaults to merge, replace rewrites this view's custom button bindings, and replace plus action_buttons=[] clears bindings without deleting button bodies", "publish=false with action_buttons is blocked as VIEW_ACTION_BUTTONS_REQUIRE_PUBLISH because the underlying custom button writer may publish after successful button writes", "legacy upsert_views[].buttons is a compatibility path for raw full replacement of existing view button config; do not use it as the normal agent path", "upsert_views[].visibility may set per-view visibility; omit it to preserve an existing view's auth or default a new view to workspace/not", "filters are saved fixed filters that apply when the view opens; query_conditions configure the frontend query panel and only apply after a user enters query values", "upsert_views[].query_conditions.rows is a layout matrix of query-panel supported field names; it is compiled to backend queryCondition queIds", "do not put relation/attachment/subtable/code_block/q_linker/address fields in query_conditions; use filters for fixed filters or app_associated_resources_apply.match_mappings for current-record relation/report matching", "use patch_views for partial parameter replacement on existing views; the public update mode is patch even though the backend save is still a full view payload", "remove_views accepts a raw view_key or an exact unique view name; after DELETE the tool verifies deletion by single view_key readback, not by a full app view list", "deleted views return verification.by_view[].delete_executed, readback_status, and safe_to_retry_delete=false; if readback is pending, do not blindly repeat the delete", "do not create business views named 全部数据, 我的数据, 我发起的, 待办, 已办, or 抄送; these are built-in system/default views. Use business-specific names for new views, and pass the existing raw view_key or patch_views when changing a built-in view", "new views created by app_views_apply default associated report/view display to visible with limit_type=all; existing views preserve their current associated display unless patch_views/upsert_views explicitly changes associated_resources", "associated report/view resource pool and per-view selected resources are configured through app_associated_resources_apply; app_views_apply keeps legacy associated_resources input compatible but it is no longer the recommended public contract", "for multi-value operators such as in, pass values as a list; value may also be used as an alias when it already contains a list", *_VISIBILITY_EXECUTION_NOTES, ], "minimal_example": { "profile": "default", "publish": True, "views": [{"operation": "upsert", "app_key": "APP_KEY", "name": "项目台账视图", "type": "table", "columns": ["项目名称"], "visibility": deepcopy(_VISIBILITY_WORKSPACE_EXAMPLE)}], }, "query_conditions_example": { "profile": "default", "app_key": "APP_KEY", "publish": True, "patch_views": [ { "view_key": "VIEW_KEY", "set": { "query_conditions": { "enabled": True, "rows": [["客户名称", "负责人"], ["创建时间"]], } }, } ], "remove_views": [], }, "action_buttons_example": { "profile": "default", "app_key": "APP_KEY", "publish": True, "upsert_views": [ { "name": "生产工单执行视图", "type": "table", "columns": ["工单编号", "产品", "状态", "负责人"], "action_buttons": [ { "text": "开始生产", "action": "link", "url": "https://example.com/start", "placement": "list", "visible_when": [{"field_name": "状态", "operator": "eq", "value": "待排产"}], } ], } ], "remove_views": [], }, "action_buttons_patch_example": { "profile": "default", "app_key": "APP_KEY", "publish": True, "patch_views": [ { "view_key": "RAW_VIEW_KEY", "set": { "action_buttons": [ { "text": "创建验收单", "action": "add_data", "target_app_key": "ARRIVAL_APP", "field_mappings": [{"source_field": "数据ID", "target_field": "关联工单"}], "default_values": {"状态": "待验收"}, "placement": "detail", } ] }, } ], "remove_views": [], }, "full_upsert_query_conditions_example": { "profile": "default", "app_key": "APP_KEY", "publish": True, "upsert_views": [ { "name": "客户查询视图", "type": "table", "columns": ["客户名称", "负责人", "客户状态", "创建时间"], "filters": [{"field_name": "客户状态", "operator": "eq", "value": "有效"}], "query_conditions": { "enabled": True, "exact": False, "hide_before_query": False, "rows": [["客户名称", "负责人"], ["创建时间"]], }, } ], "remove_views": [], }, "query_conditions_field_rules": { "supported_field_types": ["text", "long_text", "number", "amount", "date", "datetime", "single_select", "multi_select", "phone", "email", "boolean", "member", "department"], "unsupported_field_types": ["relation", "attachment", "subtable", "address", "code_block", "q_linker"], "use_instead": { "fixed_filter": "filters", "current_record_related_report_or_view": "app_associated_resources_apply.match_mappings", }, }, "gantt_example": { "profile": "default", "app_key": "APP_KEY", "publish": True, "upsert_views": [ { "name": "项目甘特图", "type": "gantt", "columns": ["项目名称", "开始日期", "结束日期"], "start_field": "开始日期", "end_field": "结束日期", "title_field": "项目名称", "filters": [{"field_name": "状态", "operator": "eq", "value": "进行中"}], } ], "remove_views": [], }, }, "app_get": { "allowed_keys": ["app_key"], "aliases": {}, "allowed_values": {}, "execution_notes": [ "returns builder-side app map: base summary, editability, field/view/chart/button counts, compact views, compact charts, custom_buttons, and app-level associated_resources", "use this as the default builder discovery read before view_get/chart_get/apply detail work", "editability is route-aware builder capability summary, not end-user data visibility", "can_edit_app_base covers app base-info writes such as app_name, icon, and visibility; it follows backend EditAppAuth and does not require package edit_tag", "can_edit_form covers form/schema routes and also follows backend EditAppAuth", "returns normalized app visibility when backend auth is readable", "custom_buttons[].button_id is the id required by app_custom_buttons_apply view_configs[].buttons[].button_ref", "associated_resources[].associated_item_id is the internal id; app_associated_resources_apply.view_configs/remove/reorder may also pass an existing resource's chart_id/chart_key/view_key", ], "minimal_example": { "profile": "default", "app_key": "APP_KEY", }, }, "app_get_fields": { "allowed_keys": ["app_key"], "aliases": {}, "allowed_values": {}, "execution_notes": [ "returns compact current field configuration for one app", "use this before app_schema_apply when you need exact field definitions", "also returns chart_fields from QingBI datasource fields; app_charts_apply field selectors should use chart_fields because record/schema-visible fields and QingBI fields are not the same schema", "chart_fields[].field_id supports field_ selectors, while chart_fields[].bi_field_id is the raw QingBI fieldId accepted by report configs", "chart_fields[].chart_apply_examples contains copyable semantic app_charts_apply snippets such as count_by_field, filtered_count, and numeric sum_metric", "subtable fields include nested subfields using the same compact field shape", ], "minimal_example": { "profile": "default", "app_key": "APP_KEY", }, }, "app_get_layout": { "allowed_keys": ["app_key"], "aliases": {}, "allowed_values": {}, "execution_notes": [ "returns compact current layout configuration for one app", "use this before app_layout_apply when you need paragraph and row structure", ], "minimal_example": { "profile": "default", "app_key": "APP_KEY", }, }, "app_get_views": { "allowed_keys": ["app_key"], "aliases": {}, "allowed_values": {}, "execution_notes": [ "returns compact current view inventory for one app", "compatibility/specialized inventory tool; default builder discovery should start with app_get", "use this before app_views_apply only when you need an exact current view inventory beyond app_get", "view items include visibility_summary when backend view auth is readable", ], "minimal_example": { "profile": "default", "app_key": "APP_KEY", }, }, "app_get_flow": { "allowed_keys": ["app_key"], "aliases": {}, "allowed_values": {}, "execution_notes": [ "returns workflow configuration summary for one app", "use this before app_flow_apply when you need the current node structure", ], "minimal_example": { "profile": "default", "app_key": "APP_KEY", }, }, "app_get_charts": { "allowed_keys": ["app_key"], "aliases": {}, "allowed_values": {}, "execution_notes": [ "returns a compact current chart inventory for one app", "compatibility/specialized inventory tool; default builder discovery should start with app_get", "use this before app_charts_apply when you need exact current chart_id values beyond the app_get summary", "chart summaries do not include full qingbi config payloads", "chart items include visibility_summary when QingBI base info is readable", ], "minimal_example": { "profile": "default", "app_key": "APP_KEY", }, }, "portal_list": { "allowed_keys": [], "aliases": {}, "allowed_values": {}, "execution_notes": [ "returns builder-configurable portal list items only", "use this as the builder portal discovery path before portal_get", "results are compact list items, not raw dash payloads", "large workspaces may return an unfiltered discovery list with portal_permissions_verified=false instead of probing every portal detail", ], "minimal_example": { "profile": "default", }, }, "portal_get": { "allowed_keys": ["dash_key", "being_draft"], "aliases": {"beingDraft": "being_draft"}, "allowed_values": {}, "execution_notes": [ "returns builder-side portal configuration detail plus a normalized component inventory", "chart and view components are returned as refs only; use builder chart_get or builder view_get for configuration detail", "being_draft=true reads the current draft view; being_draft=false reads live", "returns normalized portal visibility when backend auth is readable", ], "minimal_example": { "profile": "default", "dash_key": "DASH_KEY", "being_draft": True, }, }, "app_charts_apply": { "allowed_keys": [ "app_key", "upsert_charts", "patch_charts", "remove_chart_ids", "reorder_chart_ids", "upsert_charts[].metric", "upsert_charts[].metrics", "upsert_charts[].group_by", "upsert_charts[].where", "upsert_charts[].visibility", "patch_charts[].chart_id", "patch_charts[].name", "patch_charts[].set", "patch_charts[].unset", ], "aliases": { "patchCharts": "patch_charts", "chart.id": "chart.chart_id", "chart.type": "chart.chart_type", "chart.dimension_fields": "chart.dimension_field_ids", "chart.indicator_fields": "chart.indicator_field_ids", "chart.metric_field_ids": "chart.indicator_field_ids", "chart.dimensions": "chart.group_by", "chart.groupBy": "chart.group_by", "chart.where": "chart.filters", "chart.metric.operation": "chart.metric.op", "chart.metric.aggregation": "chart.metric.op", "chart.filter.op": "chart.filter.operator", }, "allowed_values": { "chart.chart_type": [member.value for member in PublicChartType], "chart.filter.operator": [member.value for member in ViewFilterOperator], **deepcopy(_VISIBILITY_ALLOWED_VALUES), }, "execution_notes": [ "this tool manages QingBI report bodies/configs; it does not attach reports to Qingflow app associated-resource display", "app_charts_apply creates/updates app-source QingBI reports only; generated payloads use dataSourceType=qingflow", "dataset BI reports are not created or edited by this tool yet; create them in QingBI first, then attach the existing report with app_associated_resources_apply report_source=dataset", "after creating or updating an app-source report body, use app_associated_resources_apply when the report should appear inside a Qingflow app/view", "app_charts_apply is immediate-live and does not publish", "use patch_charts for partial parameter replacement on existing charts; the tool reads current chart base/config, merges patch_charts[].set/unset, then submits full QingBI base/config payloads internally", "chart matching precedence is chart_id first, then exact unique chart name", "when chart names are not unique, supply chart_id instead of guessing by name", "successful create results must return a real backend chart_id", "upsert_charts[].visibility compiles to QingBI base visibleAuth only", "visibility-only updates keep the existing chart config and do not rewrite rawDataConfigDTO.authInfo", "chart dimension/metric/filter/query fields are resolved from app_get_fields.chart_fields (QingBI datasource fields), not record schema or form-only fields", "preferred chart metric DSL is SQL-like: metric='count(*)', metric='sum(金额)', metrics=['sum(金额)'], group_by=['状态'], where=[{field, op, value}]", "legacy dimension_field_ids/indicator_field_ids/config.aggregate remain supported as advanced compatibility input, but should not be the first choice for agents", "chart_get returns semantic group_by and metrics; raw QingBI config is diagnostic detail", "system fields such as 申请人/申请时间/编号 are usable only when they appear in chart_fields; otherwise app_charts_apply returns CHART_FIELD_NOT_IN_QINGBI_SCHEMA", "low-frequency chart types have local prevalidation: gauge requires 0 dimensions and 2 non-duplicated metrics; histogram requires at most 1 dimension and exactly 1 plain numeric metric", "chart rule failures return chart_results[].diagnostics with rule_code, expected, actual, offending_fields, and next_action; backend 81002/81005 are translated when possible", "remove_chart_ids deletes by chart_id and verifies each deleted chart with single chart_id readback; pure delete does not read the full chart list", "if delete readback is unavailable or still finds the chart, chart_results[] returns delete_executed=true, readback_status, and safe_to_retry_delete=false; do not blindly repeat delete", *_VISIBILITY_EXECUTION_NOTES, ], "minimal_example": { "profile": "default", "app_key": "APP_KEY", "upsert_charts": [{"name": "数据总量", "chart_type": "target", "metric": "count(*)", "visibility": deepcopy(_VISIBILITY_WORKSPACE_EXAMPLE)}], "patch_charts": [{"chart_id": "CHART_ID", "set": {"name": "数据总量-新版"}}], "remove_chart_ids": [], "reorder_chart_ids": [], }, }, "view_get": { "allowed_keys": ["view_key"], "aliases": {}, "allowed_values": {}, "execution_notes": [ "returns one builder-side view definition detail", "does not return record data; use user-side view_get or record_list for runtime rows", "view_key is a raw builder view key; if a record-data custom:VIEW_KEY value is passed, the tool normalizes it to VIEW_KEY", "use this after builder portal_get when a component references a view_ref.view_key", "returns normalized view visibility when backend auth is readable", ], "minimal_example": { "profile": "default", "view_key": "VIEW_KEY", }, }, "chart_get": { "allowed_keys": ["chart_id"], "aliases": {}, "allowed_values": {}, "execution_notes": [ "returns builder-side chart base info and chart config only", "chart_id is required; chart names are not accepted here", "does not return chart data; use user-side chart_get for runtime data access", "returns normalized chart visibility from QingBI visibleAuth when readable", ], "minimal_example": { "profile": "default", "chart_id": "CHART_ID", }, }, "portal_apply": { "allowed_keys": ["dash_key", "dash_name", "name", "package_id", "publish", "sections", "patch_sections", "pages", "payload", "layout_preset", "visibility", "auth", "icon", "color", "hide_copyright", "dash_global_config", "config"], "aliases": { "packageId": "package_id", "name": "dash_name", "sourceType": "source_type", "chartRef": "chart_ref", "inlineChart": "chart", "createChart": "chart", "viewRef": "view_ref", "dashStyleConfigBO": "dash_style_config", }, "section_allowed_keys": [ "title", "source_type", "role", "position", "dash_style_config", "config", "chart", "chart.app_key", "chart.name", "chart.chart_name", "chart.chart_type", "chart.metric", "chart.metrics", "chart.group_by", "chart.where", "chart.filters", "chart.chart_id", "chart.chart_key", "chart_ref", "view_ref", "text", "url", ], "section_aliases": { "sourceType": "source_type", "zone": "role", "sectionRole": "role", "chartRef": "chart_ref", "inlineChart": "chart", "createChart": "chart", "viewRef": "view_ref", "dashStyleConfigBO": "dash_style_config", }, "allowed_values": { "section.source_type": ["chart", "view", "grid", "filter", "text", "link"], "workspace_icon.icon_names": list(WORKSPACE_ICON_NAMES), "workspace_icon.icon_colors": list(WORKSPACE_ICON_COLORS), "workspace_icon.generic_icon_names": list(GENERIC_WORKSPACE_ICON_NAMES), **deepcopy(_VISIBILITY_ALLOWED_VALUES), }, "execution_notes": [ "use exactly one resource mode", "update mode: dash_key", "create mode: package_id + dash_name", "create mode follows backend DashCtrl.createDash: package add_app permission is checked on the target package; package edit_app is not required for the create precheck", "create mode requires explicit icon + color; icon=template is blocked because it is too generic", "edit mode preserves existing icon/color when omitted; explicit icon/color values are still validated", "call workspace_icon_catalog_get or `qingflow --json builder icon catalog` for supported icon/color candidates", "portal_apply uses replace semantics for sections", "when editing an existing portal, sections may be omitted to update only base info such as visibility, icon, or package", "portal section-level patch is not exposed; supplying sections means full sections replacement", "remove a section by omitting it from the new sections list", "package_id is required when creating a new portal", "publish=false only guarantees draft and base-info updates; it does not claim live has changed", "sections[].chart is the canonical QingBI chart field: chart_id/chart_key/chart_name references existing charts, while app_key+name/chart_type creates or updates app-source charts inline", f"portal inline chart creation runs as chart jobs with an internal parallel limit of {PORTAL_INLINE_CHART_MAX_WORKERS}; identical same-app chart definitions are reused, conflicting duplicates are rejected", "chart_ref and inline_chart remain compatibility aliases; do not use them as the normal agent path", "view_ref resolves by view_key first, then exact unique view_name", "pc layout uses a 24-column grid; mobile layout uses a 6-column grid", "if unsure about layout, omit position or use layout_preset=auto/dashboard_2col/dashboard_3col", "two-column pc layout should use x=0/12 with cols=12; three-column pc layout should use x=0/8/16 with cols=8", "x=0/6 with cols=6 only occupies the left half of the pc portal and triggers PORTAL_LAYOUT_HALF_WIDTH", "grid sections must include config.items; an empty grid triggers PORTAL_GRID_ITEMS_EMPTY because the frontend only shows an empty entry container", "metric portal sections may set role=metric; role=metric requires a target/indicator chart and otherwise triggers PORTAL_METRIC_SECTION_CHART_TYPE_MISMATCH", "standard workbench count diagnostics warn when metric cards are not 4-6, BI charts are not 2-3, or business views are not 1-2", "position.pc/mobile is the canonical portal layout shape", "compat payload accepts name -> dash_name and single pages[0].components -> sections", "visibility is the canonical public auth shape; auth is kept only as a deprecated compatibility alias", "passing visibility and auth together is rejected as VISIBILITY_CONFLICT", *_VISIBILITY_EXECUTION_NOTES, ], "minimal_example": { "profile": "default", "dash_name": "经营门户", "package_id": 1001, "icon": "view-grid", "color": "blue", "publish": True, "layout_preset": "dashboard_2col", "visibility": deepcopy(_VISIBILITY_WORKSPACE_EXAMPLE), "sections": [ { "title": "经营概览", "source_type": "chart", "chart": { "app_key": "APP_KEY", "name": "数据总量", "chart_type": "summary", "metric": {"field": "数据ID", "agg": "count"}, }, "position": { "pc": {"x": 0, "y": 0, "cols": 12, "rows": 6}, "mobile": {"x": 0, "y": 0, "cols": 6, "rows": 6}, }, } ], }, "compat_payload_example": { "name": "经营门户", "package_id": 1001, "layout_preset": "dashboard_2col", "pages": [ { "title": "经营总览", "components": [ { "title": "销售趋势", "source_type": "chart", "chart": {"chart_id": "CHART_ID"}, } ], } ], }, "minimal_section_example": { "title": "订单概览", "source_type": "view", "view_ref": {"app_key": "APP_KEY", "view_key": "VIEW_KEY"}, "position": { "pc": {"x": 0, "y": 0, "cols": 24, "rows": 8}, "mobile": {"x": 0, "y": 0, "cols": 6, "rows": 8}, }, }, }, "portal_delete": { "allowed_keys": ["dash_key"], "aliases": {"dashKey": "dash_key"}, "allowed_values": {}, "execution_notes": [ "deletes one portal by dash_key", "delete results separate DELETE execution from readback verification", "if delete_executed=true and readback_status=unavailable or still_exists, do not blindly repeat delete; confirm later with portal_get or portal_list", ], "minimal_example": { "profile": "default", "dash_key": "DASH_KEY", }, }, "app_publish_verify": { "allowed_keys": ["app_key", "expected_package_id"], "aliases": {}, "allowed_values": {}, "execution_notes": [ "verifies that the current app draft has been published and is readable through the builder surface", "expected_package_id is optional and adds an extra package consistency check", "when verification fails because of an edit lock owned by the current user, retry after app_release_edit_lock_if_mine", ], "minimal_example": { "profile": "default", "app_key": "APP_KEY", "expected_package_id": 1001, }, }, "app_repair_code_blocks": { "allowed_keys": ["app_key", "field", "apply"], "aliases": {}, "allowed_values": {}, "execution_notes": [ "scan existing code_block fields for qf_output shadowing, missing output assignment, and broken writeback defaults", "apply=false is a dry-run and returns a repair plan without writing builder config", "apply=true rewrites only safe cases: const/let qf_output shadowing and stale relation-default output bindings reconstructed from readable code_block_binding config", "this tool does not invent missing output aliases or guess missing target bindings when only low-level codeBlockConfig is present", ], "minimal_example": { "profile": "default", "app_key": "APP_KEY", "field": "线索价值评估", "apply": False, }, }, } _PRIVATE_BUILDER_TOOL_CONTRACTS = { "app_schema_plan", "app_layout_plan", "app_flow_plan", "app_views_plan", } _BUILDER_TOOL_CONTRACT_ALIASES = {}