from __future__ import annotations import json from copy import deepcopy from functools import wraps from typing import Any, Callable from .public_surface import ( BUILDER_DOMAIN, USER_DOMAIN, cli_trim_key_from_namespace, server_method_map, tool_key, ) JSONObject = dict[str, Any] TransformFn = Callable[[JSONObject], None] COMMON_SUCCESS_DROP_TOP = { "request_route", "profile", "ws_id", "output_profile", "qf_version_source", "persisted", "base_url", "normalized_args", "suggested_next_call", "noop", "allowed_values", "missing_fields", "ok", } COMMON_ERROR_DROP_TOP = { "request_route", "base_url", "normalized_args", "suggested_next_call", "profile", "ws_id", "output_profile", "qf_version_source", "persisted", "allowed_values", "missing_fields", "noop", } SUCCESS_POLICY_BY_TOOL: dict[str, TransformFn] = {} def trim_public_response(tool_name: str | None, payload: dict[str, Any]) -> dict[str, Any]: if not isinstance(payload, dict): return payload if _looks_like_failure_payload(payload): status = str(payload.get("status") or "").lower() if _is_executed_nonfatal_payload(payload) or ( tool_name in {"user:record_insert", "user:record_update", "user:record_delete"} and status in { "blocked", "needs_confirmation", "partial_success", } ) or ( tool_name == "user:record_import_verify" and status == "failed" and "can_import" in payload ): return trim_success_response(tool_name, payload) return _trim_returned_failure(payload) return trim_success_response(tool_name, payload) def trim_success_response(tool_name: str | None, payload: dict[str, Any]) -> dict[str, Any]: if not isinstance(payload, dict): return payload trimmed = deepcopy(payload) drop_keys = COMMON_SUCCESS_DROP_TOP if tool_name in {"user:record_get", "user:record_logs_get"}: drop_keys = COMMON_SUCCESS_DROP_TOP - {"output_profile"} if tool_name in {"user:record_insert", "user:record_update", "user:record_delete"} and payload.get("ok") is False: drop_keys = drop_keys - {"ok"} _drop_top_keys(trimmed, drop_keys) transformer = SUCCESS_POLICY_BY_TOOL.get(tool_name or "") if transformer is not None: transformer(trimmed) _drop_empty_optional_keys(trimmed) return trimmed def trim_error_response(payload: dict[str, Any]) -> dict[str, Any]: if not isinstance(payload, dict): return payload trimmed = deepcopy(payload) _drop_top_keys(trimmed, COMMON_ERROR_DROP_TOP) _drop_deep_keys(trimmed.get("details"), {"request_route", "base_url", "normalized_args", "suggested_next_call", "transport", "response", "body", "raw"}) details = trimmed.get("details") if isinstance(details, dict): preserved = {} for key in ("blocking_issues", "compiled_match_rules", "issues", "expected_shape"): if key in details: preserved[key] = details.get(key) compact_details = _compact_scalar_dict(details) compact_details.update(preserved) if compact_details: trimmed["details"] = compact_details else: trimmed.pop("details", None) if trimmed.get("backend_code") is not None: trimmed.pop("http_status", None) _drop_empty_optional_keys(trimmed) return trimmed def wrap_trimmed_methods(instance: object, method_map: dict[str, str]) -> object: for method_name, tool_name in method_map.items(): original = getattr(instance, method_name, None) if original is None or not callable(original): continue setattr(instance, method_name, _wrap_callable(original, tool_name)) return instance def resolve_cli_tool_name(args: Any) -> str | None: return cli_trim_key_from_namespace(args) USER_SERVER_METHOD_MAP = server_method_map(USER_DOMAIN) BUILDER_SERVER_METHOD_MAP = server_method_map(BUILDER_DOMAIN) def _wrap_callable(original: Callable[..., Any], tool_name: str) -> Callable[..., Any]: @wraps(original) def wrapped(*args: Any, **kwargs: Any) -> Any: try: result = original(*args, **kwargs) except RuntimeError as exc: payload = _parse_runtime_error_payload(exc) if payload is None: raise raise RuntimeError(json.dumps(trim_error_response(payload), ensure_ascii=False)) from None return trim_public_response(tool_name, result) if isinstance(result, dict) else result return wrapped def _parse_runtime_error_payload(exc: RuntimeError) -> dict[str, Any] | None: raw = str(exc) try: payload = json.loads(raw) except json.JSONDecodeError: return None return payload if isinstance(payload, dict) else None def _looks_like_failure_payload(payload: dict[str, Any]) -> bool: if _is_executed_nonfatal_payload(payload): return False if payload.get("ok") is False: return True status = str(payload.get("status") or "").lower() return status in {"failed", "blocked"} def _is_executed_nonfatal_payload(payload: dict[str, Any]) -> bool: status = str(payload.get("status") or "").lower() executed = bool( payload.get("write_executed") or payload.get("write_may_have_succeeded") or payload.get("delete_executed") or payload.get("action_executed") or payload.get("export_executed") ) return executed and status in {"partial_success", "verification_failed", "running", "queued", "unknown"} def _trim_returned_failure(payload: dict[str, Any]) -> dict[str, Any]: trimmed = deepcopy(payload) _drop_top_keys(trimmed, COMMON_ERROR_DROP_TOP) _drop_deep_keys(trimmed.get("details"), {"request_route", "base_url", "normalized_args", "suggested_next_call", "transport", "response", "body", "raw"}) details = trimmed.get("details") if isinstance(details, dict): preserved = {} for key in ("blocking_issues", "compiled_match_rules", "issues", "expected_shape"): if key in details: preserved[key] = details.get(key) compact_details = _compact_scalar_dict(details) compact_details.update(preserved) if compact_details: trimmed["details"] = compact_details else: trimmed.pop("details", None) if trimmed.get("backend_code") is not None: trimmed.pop("http_status", None) _drop_empty_optional_keys(trimmed) return trimmed def _drop_top_keys(payload: JSONObject, keys: set[str]) -> None: for key in keys: payload.pop(key, None) def _drop_empty_optional_keys(payload: JSONObject) -> None: for key in ("warnings", "verification", "details"): value = payload.get(key) if value in (None, [], {}, ""): payload.pop(key, None) def _compact_scalar_dict(payload: dict[str, Any]) -> dict[str, Any]: compact: dict[str, Any] = {} for key, value in payload.items(): if isinstance(value, (str, int, float, bool)) or value is None: compact[key] = value continue if isinstance(value, dict): nested = { nested_key: nested_value for nested_key, nested_value in value.items() if isinstance(nested_value, (str, int, float, bool)) or nested_value is None } if nested: compact[key] = nested return compact def _trim_item_list(payload: JSONObject, key: str, *, allowed: tuple[str, ...]) -> None: items = payload.get(key) if not isinstance(items, list): return payload[key] = [_pick(item, allowed) for item in items if isinstance(item, dict)] def _trim_nested_item_list(payload: JSONObject, path: tuple[str, ...], *, allowed: tuple[str, ...]) -> None: parent = _nested_dict(payload, path[:-1]) if not isinstance(parent, dict): return key = path[-1] items = parent.get(key) if not isinstance(items, list): return parent[key] = [_pick(item, allowed) for item in items if isinstance(item, dict)] def _keep_nested_keys(payload: JSONObject, path: tuple[str, ...], *, allowed: tuple[str, ...]) -> None: parent = _nested_dict(payload, path[:-1]) if not isinstance(parent, dict): return node = parent.get(path[-1]) if isinstance(node, dict): parent[path[-1]] = _pick(node, allowed) def _drop_nested_keys(payload: JSONObject, path: tuple[str, ...], *, keys: tuple[str, ...]) -> None: parent = _nested_dict(payload, path[:-1]) if not isinstance(parent, dict): return node = parent.get(path[-1]) if not isinstance(node, dict): return for key in keys: node.pop(key, None) def _drop_deep_keys(payload: Any, keys: set[str]) -> None: if isinstance(payload, dict): for key in list(payload.keys()): if key in keys: payload.pop(key, None) continue _drop_deep_keys(payload.get(key), keys) elif isinstance(payload, list): for item in payload: _drop_deep_keys(item, keys) def _nested_dict(payload: JSONObject, path: tuple[str, ...]) -> JSONObject | None: node: Any = payload for key in path: if not isinstance(node, dict): return None node = node.get(key) return node if isinstance(node, dict) else None def _pick(payload: JSONObject, allowed: tuple[str, ...]) -> JSONObject: return {key: payload.get(key) for key in allowed if key in payload} def _trim_auth_payload(payload: JSONObject) -> None: pass def _trim_auth_logout(payload: JSONObject) -> None: pass def _trim_workspace_list(payload: JSONObject) -> None: page = payload.get("page") if isinstance(page, dict): _trim_item_list(page, "list", allowed=("wsId", "workspaceName", "remark")) def _trim_workspace_get(payload: JSONObject) -> None: workspace = payload.get("workspace") if isinstance(workspace, dict): payload["workspace"] = _pick( workspace, allowed=("wsId", "workspaceName", "remark", "systemVersion", "auth"), ) def _trim_app_list_like(payload: JSONObject) -> None: payload.pop("apps", None) _trim_item_list(payload, "items", allowed=("app_key", "app_name", "package_name")) def _trim_app_get(payload: JSONObject) -> None: _trim_builder_envelope(payload) _trim_nested_item_list( payload, ("data", "accessible_views"), allowed=("view_id", "name", "analysis_supported", "list_supported", "kind"), ) def _trim_file_upload_info(payload: JSONObject) -> None: pass def _trim_file_upload_local(payload: JSONObject) -> None: payload.pop("result", None) payload.pop("upload_result", None) def _trim_import_schema(payload: JSONObject) -> None: columns: list[JSONObject] | None = None if isinstance(payload.get("columns"), list): columns = [item for item in payload.get("columns", []) if isinstance(item, dict)] elif isinstance(payload.get("expected_columns"), list): columns = [item for item in payload.get("expected_columns", []) if isinstance(item, dict)] if columns is not None: payload["columns"] = [_compact_import_column(item) for item in columns] import_capability = payload.get("import_capability") if isinstance(payload.get("import_capability"), dict) else {} if import_capability: verification = payload.get("verification") if isinstance(payload.get("verification"), dict) else {} if not isinstance(payload.get("verification"), dict): payload["verification"] = verification verification.setdefault("import_auth_source", import_capability.get("auth_source")) verification.setdefault("import_capability_can_import", import_capability.get("can_import")) payload.pop("expected_columns", None) payload.pop("schema_fingerprint", None) payload.pop("import_capability", None) payload.pop("request_route", None) if _looks_like_import_verify(payload): _trim_import_verify_payload(payload) return payload.pop("verification", None) if "applied_repairs" in payload or "repaired_file_path" in payload: _trim_import_repair_payload(payload) return if "template_url" in payload or "downloaded_to_path" in payload: _trim_import_template_payload(payload) return if "import_id" in payload or "process_id_str" in payload: _trim_import_status_payload(payload) return def _trim_record_schema(payload: JSONObject) -> None: payload.pop("legacy_schema", None) template_map = payload.get("payload_template") if not isinstance(template_map, dict): template_map = None if "writable_fields" in payload: writable_fields = payload.get("writable_fields") payload.pop("writable_fields", None) required_fields: list[JSONObject] = [] optional_fields: list[JSONObject] = [] if isinstance(writable_fields, list): for item in writable_fields: compact = _compact_schema_field(item, template_map=template_map) if not compact: continue if compact.get("required") is True: required_fields.append(compact) else: optional_fields.append(compact) payload["required_fields"] = required_fields payload["optional_fields"] = optional_fields for key in ("required_fields", "optional_fields", "runtime_linked_required_fields", "fields", "ambiguous_fields"): if key in payload: payload[key] = _compact_schema_fields(payload.get(key), template_map=template_map) for key in ("suggested_dimensions", "suggested_metrics", "suggested_time_fields"): if isinstance(payload.get(key), list): payload[key] = [ _pick(item, ("field_id", "title")) for item in payload.get(key) if isinstance(item, dict) ] for key in ("workflow_node", "view_resolution", "field_count"): payload.pop(key, None) def _trim_record_write(payload: JSONObject) -> None: payload.pop("verification", None) data = payload.get("data") if not isinstance(data, dict): return if payload.get("mode") == "batch" or data.get("mode") == "batch": _trim_record_write_batch(payload, data) return data.pop("debug", None) data.pop("normalized_payload", None) data.pop("human_review", None) data.pop("action", None) if payload.get("status") == "success": data.pop("tried_routes", None) update_route = payload.get("update_route") if update_route not in (None, [], {}, ""): data["update_route"] = update_route tried_routes = payload.get("tried_routes") if payload.get("status") != "success" and tried_routes not in (None, [], {}, ""): data["tried_routes"] = tried_routes resource = _compact_record_resource(data.get("resource")) if resource: data["resource"] = resource else: data.pop("resource", None) verification = data.get("verification") if isinstance(verification, dict): compact_verification = _pick( verification, ( "verified", "verification_mode", "field_level_verified", ), ) if compact_verification: data["verification"] = compact_verification else: data.pop("verification", None) for key in ("blockers", "field_errors", "confirmation_requests", "resolved_fields"): value = data.get(key) if value in (None, [], {}, ""): data.pop(key, None) def _trim_record_write_batch(payload: JSONObject, data: JSONObject) -> None: data.pop("items", None) data.pop("debug", None) for key in ("summary", "app_key", "mode"): if data.get(key) in (None, [], {}, ""): data.pop(key, None) items = payload.get("items") if isinstance(items, list): payload["items"] = [ _pick( item, ( "index", "row_number", "status", "record_id", "apply_id", "write_executed", "write_succeeded", "verification_status", "safe_to_retry", "update_route", "tried_routes", "failed_fields", "confirmation_requests", "blockers", "error", "warnings", "resource", "verification", ), ) for item in items if isinstance(item, dict) ] if payload.get("items") in (None, [], {}, ""): payload.pop("items", None) def _trim_record_get(payload: JSONObject) -> None: if payload.get("fields") is not None or payload.get("semantic_context") is not None: _trim_detail_context_record_get(payload) return data = payload.get("data") if not isinstance(data, dict): return compact: dict[str, Any] = {} app_key = data.get("app_key") if app_key: compact["app_key"] = app_key record_id = data.get("record_id") if record_id not in (None, ""): compact["record_id"] = str(record_id) record = data.get("record") if isinstance(record, dict): compact["record"] = record normalized_record = data.get("normalized_record") if isinstance(normalized_record, dict): compact["normalized_record"] = normalized_record normalized_ambiguous_fields = data.get("normalized_ambiguous_fields") if isinstance(normalized_ambiguous_fields, dict): compact["normalized_ambiguous_fields"] = normalized_ambiguous_fields payload["data"] = compact def _trim_detail_context_record_get(payload: JSONObject) -> None: _trim_item_list( payload, "fields", allowed=("field_id", "title", "type", "value", "display_value", "requested_focus", "asset_ids", "file_asset_ids"), ) references = payload.get("references") if isinstance(references, list): compact_refs: list[JSONObject] = [] for item in references: if not isinstance(item, dict): continue compact = _pick( item, ( "field_id", "field_title", "target_app_key", "target_record_id", "target_title", "target_detail_completeness", "self_reference", ), ) target_fields = item.get("target_fields") if isinstance(item.get("target_fields"), list) else [] if target_fields: compact["target_fields"] = [ _pick(field, ("field_id", "title", "type", "value", "display_value", "asset_ids", "file_asset_ids")) for field in target_fields if isinstance(field, dict) ] compact_refs.append(compact) payload["references"] = compact_refs media_assets = payload.get("media_assets") if isinstance(media_assets, dict): compact_media = _pick(media_assets, ("status", "local_dir", "warnings")) items = media_assets.get("items") if isinstance(media_assets.get("items"), list) else [] compact_media["items"] = [ _pick( item, ( "asset_id", "kind", "source", "field_id", "field_title", "local_path", "access_status", "readable_by_agent", "mime_type", "size_bytes", "download_strategy", "storage_auth_type", "storage_cookie_prefix", "redirected", ), ) for item in items if isinstance(item, dict) ] payload["media_assets"] = compact_media file_assets = payload.get("file_assets") if isinstance(file_assets, dict): compact_files = _pick(file_assets, ("status", "local_dir", "warnings")) items = file_assets.get("items") if isinstance(file_assets.get("items"), list) else [] compact_files["items"] = [ _pick( item, ( "file_asset_id", "media_asset_id", "kind", "source", "field_id", "field_title", "file_name", "local_path", "access_status", "readable_by_agent", "mime_type", "size_bytes", "download_strategy", "storage_auth_type", "storage_cookie_prefix", "redirected", "extraction", ), ) for item in items if isinstance(item, dict) ] payload["file_assets"] = compact_files for key in ("data_logs", "workflow_logs"): node = payload.get(key) if not isinstance(node, dict): continue compact_node = _pick( node, ( "status", "visible", "reason", "page", "page_size", "items_loaded", "reported_total", "has_more", "complete", "items", ), ) payload[key] = compact_node associated_resources = payload.get("associated_resources") if isinstance(associated_resources, list): payload["associated_resources"] = [ _pick(item, ("type", "resource_type", "name", "app_key", "app_name", "view_key", "chart_key", "view_type", "report_source", "data_access")) for item in associated_resources if isinstance(item, dict) ] _drop_deep_keys(payload, {"raw", "debug"}) def _trim_record_list(payload: JSONObject) -> None: data = payload.get("data") if not isinstance(data, dict): return pagination = data.get("pagination") if isinstance(data.get("pagination"), dict) else {} returned_items = pagination.get("returned_count", pagination.get("returned_items")) result_amount = pagination.get("total_count", pagination.get("result_amount")) truncated = bool(pagination.get("truncated")) if not truncated and isinstance(result_amount, int) and isinstance(returned_items, int): truncated = result_amount > returned_items compact_pagination = { "loaded": True, "returned_count": returned_items, "total_count": result_amount, "truncated": truncated, } selection = data.get("selection") if isinstance(data.get("selection"), dict) else {} view = selection.get("view") if isinstance(selection.get("view"), dict) else {} compact: dict[str, Any] = { "app_key": data.get("app_key"), "items": data.get("items") if isinstance(data.get("items"), list) else [], "pagination": compact_pagination, } if view: compact["view"] = _pick(view, ("view_id", "name")) payload["data"] = compact lookup = payload.get("lookup") if isinstance(payload.get("lookup"), dict) else {} if lookup: payload["lookup"] = { "mode": lookup.get("mode"), "query": lookup.get("query"), "total_count": lookup.get("total_count", lookup.get("reported_total")), "returned_count": lookup.get("returned_count"), "truncated": lookup.get("truncated"), "confidence": lookup.get("confidence"), "next_action": lookup.get("next_action"), } def _trim_record_access(payload: JSONObject) -> None: compact: dict[str, Any] = {} for key in ( "ok", "status", "app_key", "view_id", "format", "local_dir", "row_count", "complete", "truncated", "safe_for_final_conclusion", "files", "fields", "warnings", "verification", "scope", ): value = payload.get(key) if value is not None: compact[key] = value payload.clear() payload.update(compact) def _trim_record_logs(payload: JSONObject) -> None: compact: dict[str, Any] = {} for key in ( "ok", "status", "output_profile", "app", "view", "record", "local_dir", "summary_path", "warnings", "unavailable_context", "context_integrity", ): value = payload.get(key) if value is not None: compact[key] = value for key in ("data_logs", "workflow_logs"): node = payload.get(key) if isinstance(node, dict): compact[key] = _pick( node, ( "status", "visible", "source", "reason", "complete", "items_count", "pages_fetched", "page_size", "reported_total", "local_path", "preview_items", "warnings", "stopped_reason", ), ) payload.clear() payload.update(compact) def _trim_record_analyze(payload: JSONObject) -> None: summary: dict[str, Any] = {} completeness = payload.get("completeness") if isinstance(completeness, dict): summary["completeness"] = completeness presentation = payload.get("presentation") if isinstance(presentation, dict): summary["presentation"] = presentation ranking = payload.get("ranking") if isinstance(ranking, dict): summary["ranking"] = ranking error = payload.get("error") if isinstance(error, dict): summary["error"] = error if summary: payload["summary"] = summary for key in ("query", "ranking", "ratios", "completeness", "presentation", "error", "debug"): payload.pop(key, None) def _trim_code_block_schema(payload: JSONObject) -> None: payload.pop("legacy_schema", None) _trim_nested_item_list(payload, ("code_block_fields",), allowed=("title", "selector", "bound_output_fields", "configured_aliases")) def _trim_code_block_run(payload: JSONObject) -> None: _drop_deep_keys(payload, {"debug", "request_route"}) def _trim_task_list(payload: JSONObject) -> None: _drop_nested_keys(payload, ("data", "selection"), keys=("query",)) def _trim_task_get(payload: JSONObject) -> None: _drop_deep_keys(payload, {"request_route", "output_profile"}) def _trim_task_context_detail(payload: JSONObject) -> None: _drop_deep_keys(payload, {"request_route", "output_profile"}) def _trim_record_delete(payload: JSONObject) -> None: data = payload.get("data") if not isinstance(data, dict): return resource = data.get("resource") deleted_ids = payload.get("deleted_ids") if isinstance(payload.get("deleted_ids"), list) else data.get("deleted_ids") failed_ids = payload.get("failed_ids") if isinstance(payload.get("failed_ids"), list) else data.get("failed_ids") if not isinstance(deleted_ids, list): deleted_ids = [] if not isinstance(failed_ids, list): failed_ids = [] if not deleted_ids and isinstance(resource, dict): raw_ids = resource.get("record_ids") or resource.get("apply_ids") or resource.get("applyIds") if isinstance(raw_ids, list): deleted_ids = [str(item) for item in raw_ids if item not in (None, "")] data["deleted_ids"] = [str(item) for item in deleted_ids if item not in (None, "")] data["failed_ids"] = [str(item) for item in failed_ids if item not in (None, "")] for key in ( "resource", "action", "normalized_payload", "human_review", "verification", "blockers", "field_errors", "confirmation_requests", "resolved_fields", "debug", ): data.pop(key, None) def _compact_record_resource(resource: Any) -> dict[str, Any] | None: if not isinstance(resource, dict): return None compact: dict[str, Any] = {} if resource.get("type") not in (None, ""): compact["type"] = resource.get("type") app_key = resource.get("app_key") or resource.get("appKey") if app_key not in (None, ""): compact["app_key"] = app_key record_id = resource.get("record_id") if record_id not in (None, ""): compact["record_id"] = str(record_id) apply_id = resource.get("apply_id") or resource.get("applyId") if apply_id not in (None, "") and "record_id" not in compact: compact["record_id"] = str(apply_id) record_ids = resource.get("record_ids") if isinstance(record_ids, list): compact["record_ids"] = [str(item) for item in record_ids if item not in (None, "")] apply_ids = resource.get("apply_ids") or resource.get("applyIds") if isinstance(apply_ids, list) and "record_ids" not in compact: compact["record_ids"] = [str(item) for item in apply_ids if item not in (None, "")] return compact or None def _compact_schema_fields(items: Any, *, template_map: dict[str, Any] | None) -> list[JSONObject]: if not isinstance(items, list): return [] compacted: list[JSONObject] = [] for item in items: compact = _compact_schema_field(item, template_map=template_map) if compact: compacted.append(compact) return compacted def _compact_schema_field(item: Any, *, template_map: dict[str, Any] | None) -> JSONObject | None: if not isinstance(item, dict): return None compact: dict[str, Any] = {} field_id = item.get("field_id") if field_id not in (None, ""): compact["field_id"] = field_id title = item.get("title") if title not in (None, ""): compact["title"] = title kind = item.get("kind") or item.get("write_kind") if kind not in (None, ""): compact["kind"] = kind if "required" in item: compact["required"] = bool(item.get("required")) if template_map is not None and isinstance(title, str) and title in template_map: compact["template"] = template_map.get(title) for key in ( "format_hint", "example_value", "linkage", "may_become_required", "activation_sources", "requirement_reason", "accepts_natural_input", "requires_upload", ): value = item.get(key) if value not in (None, [], {}, ""): compact[key] = value candidate_hint = item.get("candidate_hint") if isinstance(candidate_hint, dict): compact["candidate_hint"] = candidate_hint options = item.get("options") if isinstance(options, list) and options: compact["options"] = options target_app_key = item.get("target_app_key") if isinstance(target_app_key, str) and target_app_key: compact["target_app_key"] = target_app_key searchable_fields = item.get("searchable_fields") if isinstance(searchable_fields, list) and searchable_fields: compact["searchable_fields"] = searchable_fields row_fields = item.get("row_fields") if isinstance(row_fields, list) and row_fields: compact["row_fields"] = _compact_schema_fields(row_fields, template_map=None) return compact or None def _compact_import_column(item: dict[str, Any]) -> dict[str, Any]: compact: dict[str, Any] = {} title = item.get("title") if title not in (None, ""): compact["title"] = title kind = item.get("kind") or item.get("write_kind") if kind not in (None, ""): compact["kind"] = kind compact["required"] = bool(item.get("required")) options = item.get("options") if isinstance(options, list) and options: compact["options"] = options if bool(item.get("accepts_natural_input")): compact["accepts_natural_input"] = True import_value_format = item.get("import_value_format") if isinstance(import_value_format, str) and import_value_format: compact["import_value_format"] = import_value_format format_hint = item.get("format_hint") if isinstance(format_hint, str) and format_hint: compact["format_hint"] = format_hint if bool(item.get("requires_upload")): compact["requires_upload"] = True target_app_key = item.get("target_app_key") if isinstance(target_app_key, str) and target_app_key: compact["target_app_key"] = target_app_key target_app_name = item.get("target_app_name") if isinstance(target_app_name, str) and target_app_name: compact["target_app_name"] = target_app_name searchable_fields = item.get("searchable_fields") if isinstance(searchable_fields, list) and searchable_fields: compact["searchable_fields"] = searchable_fields return compact def _looks_like_import_verify(payload: JSONObject) -> bool: if "verification_id" in payload and "can_import" in payload: return True return "can_import" in payload and ( "error_code" in payload or "issues" in payload or "file_path" in payload ) def _trim_import_verify_payload(payload: JSONObject) -> None: issues = payload.get("issues") if isinstance(payload.get("issues"), list) else [] issue_summary = _summarize_import_issues(issues) payload["issue_summary"] = issue_summary verification = payload.get("verification") if isinstance(payload.get("verification"), dict) else {} compact_verification = { key: verification.get(key) for key in ( "import_auth_prechecked", "import_auth_precheck_passed", "import_auth_source", "import_capability_can_import", "local_precheck_passed", "backend_verification_passed", "auto_normalized", ) if key in verification } if compact_verification: payload["verification"] = compact_verification else: payload.pop("verification", None) columns = payload.get("columns") if "expected_columns" not in payload and isinstance(columns, list): payload["expected_columns"] = columns file_name = payload.get("file_name") if not file_name: file_path = payload.get("file_path") if isinstance(file_path, str) and file_path: payload["file_name"] = file_path.split("/")[-1] for key in ("apply_rows", "schema_fingerprint", "import_capability", "file_sha256", "verified_file_sha256", "file_format", "local_precheck_limited"): payload.pop(key, None) def _trim_import_repair_payload(payload: JSONObject) -> None: payload["verification_id"] = payload.get("new_verification_id") or payload.get("verification_id") post_repair_issues = payload.get("post_repair_issues") if isinstance(post_repair_issues, list): payload["post_repair_issue_summary"] = _summarize_import_issues(post_repair_issues) for key in ("new_verification_id", "verification"): payload.pop(key, None) def _trim_import_template_payload(payload: JSONObject) -> None: for key in ("schema_fingerprint", "verification"): payload.pop(key, None) def _trim_import_status_payload(payload: JSONObject) -> None: total_rows = payload.get("total_rows") success_rows = payload.get("success_rows") failed_rows = payload.get("failed_rows") payload["total"] = total_rows if isinstance(success_rows, int) and isinstance(failed_rows, int): payload["finished"] = success_rows + failed_rows elif isinstance(success_rows, int): payload["finished"] = success_rows else: payload["finished"] = None payload["succeeded"] = success_rows payload["failed"] = failed_rows for key in ( "matched_by", "source_file_name", "total_rows", "success_rows", "failed_rows", "operate_time", "operate_user", "verification", ): payload.pop(key, None) def _trim_export_payload(payload: JSONObject) -> None: payload.pop("backend_export_id", None) def _summarize_import_issues(issues: list[Any]) -> dict[str, Any]: total = 0 error_count = 0 warning_count = 0 sample: list[dict[str, Any]] = [] for item in issues: if not isinstance(item, dict): continue total += 1 severity = str(item.get("severity") or "").lower() if severity == "error": error_count += 1 if severity == "warning": warning_count += 1 if len(sample) < 3: sample.append( { "code": item.get("code"), "message": item.get("message"), "severity": item.get("severity"), } ) return { "total": total, "errors": error_count, "warnings": warning_count, "sample": sample, } def _trim_directory(payload: JSONObject) -> None: pass def _trim_feedback(payload: JSONObject) -> None: payload.pop("normalized_payload", None) def _trim_builder_envelope(payload: JSONObject) -> None: if str(payload.get("status") or "").lower() in {"success", "partial_success", "verification_failed"}: details = payload.get("details") if isinstance(details, dict): _drop_deep_keys(details, {"request_route", "base_url", "normalized_args", "suggested_next_call", "transport", "response", "body", "raw"}) preserved = {} for key in ("compiled_match_rules", "issues", "expected_shape"): if key in details: preserved[key] = details.get(key) compact = _compact_scalar_dict(details) compact.update(preserved) if compact: payload["details"] = compact else: payload.pop("details", None) def _trim_builder_list_like(payload: JSONObject) -> None: _trim_builder_envelope(payload) def _register_policy(domains: tuple[str, ...], names: tuple[str, ...], transform: TransformFn) -> None: for domain in domains: for name in names: SUCCESS_POLICY_BY_TOOL[tool_key(domain, name)] = transform _register_policy((USER_DOMAIN, BUILDER_DOMAIN), ("auth_use_credential", "auth_whoami"), _trim_auth_payload) _register_policy((USER_DOMAIN, BUILDER_DOMAIN), ("auth_logout",), _trim_auth_logout) _register_policy((USER_DOMAIN, BUILDER_DOMAIN), ("workspace_list",), _trim_workspace_list) _register_policy((USER_DOMAIN, BUILDER_DOMAIN), ("workspace_get", "workspace_select"), _trim_workspace_get) _register_policy((USER_DOMAIN,), ("app_list",), _trim_app_list_like) _register_policy((USER_DOMAIN, BUILDER_DOMAIN), ("app_get",), _trim_app_get) _register_policy((BUILDER_DOMAIN,), ("app_repair_code_blocks",), _trim_builder_list_like) _register_policy((USER_DOMAIN, BUILDER_DOMAIN), ("portal_list", "portal_get", "view_get", "chart_get"), _trim_builder_list_like) _register_policy((USER_DOMAIN,), ("file_get_upload_info",), _trim_file_upload_info) _register_policy((USER_DOMAIN, BUILDER_DOMAIN), ("file_upload_local",), _trim_file_upload_local) _register_policy( (USER_DOMAIN,), ( "record_import_schema_get", "record_import_template_get", "record_import_verify", "record_import_repair_local", "record_import_start", "record_import_status_get", ), _trim_import_schema, ) _register_policy( (USER_DOMAIN,), ( "record_export_start", "record_export_status_get", "record_export_get", "record_export_direct", ), _trim_export_payload, ) _register_policy( (USER_DOMAIN,), ( "record_schema_get", "record_insert_schema_get", "record_browse_schema_get", "record_update_schema_get", "record_code_block_schema_get", ), _trim_record_schema, ) _register_policy((USER_DOMAIN,), ("record_insert", "record_update"), _trim_record_write) _register_policy((USER_DOMAIN,), ("record_get",), _trim_record_get) _register_policy((USER_DOMAIN,), ("record_list",), _trim_record_list) _register_policy((USER_DOMAIN,), ("record_access",), _trim_record_access) _register_policy((USER_DOMAIN,), ("record_logs_get",), _trim_record_logs) _register_policy((USER_DOMAIN,), ("record_analyze",), _trim_record_analyze) _register_policy((USER_DOMAIN,), ("record_code_block_run",), _trim_code_block_run) _register_policy((USER_DOMAIN,), ("task_list",), _trim_task_list) _register_policy((USER_DOMAIN,), ("task_get",), _trim_task_get) _register_policy( (USER_DOMAIN,), ( "task_action_execute", "task_associated_report_detail_get", "task_workflow_log_get", ), _trim_task_context_detail, ) _register_policy( (USER_DOMAIN,), ( "directory_search", "directory_list_internal_users", "directory_list_all_internal_users", "directory_list_internal_departments", "directory_list_all_departments", "directory_list_sub_departments", "directory_list_external_members", ), _trim_directory, ) _register_policy((USER_DOMAIN, BUILDER_DOMAIN), ("feedback_submit",), _trim_feedback) _register_policy( (USER_DOMAIN,), ( "record_member_candidates", "record_department_candidates", ), _trim_builder_list_like, ) _register_policy((USER_DOMAIN,), ("record_delete",), _trim_record_delete) _register_policy( (BUILDER_DOMAIN,), ( "builder_tool_contract", "workspace_icon_catalog_get", "package_list", "package_get", "package_apply", "solution_install", "member_search", "role_search", "role_create", "app_release_edit_lock_if_mine", "app_resolve", "button_style_catalog_get", "app_custom_buttons_apply", "app_associated_resources_apply", "app_get_fields", "app_repair_code_blocks", "app_get_layout", "app_get_views", "app_get_flow", "app_get_charts", "app_schema_apply", "app_layout_apply", "app_flow_apply", "app_views_apply", "app_charts_apply", "portal_apply", "portal_delete", "app_publish_verify", ), _trim_builder_list_like, )