from __future__ import annotations import json import re from typing import Any from uuid import uuid4 from mcp.server.fastmcp import FastMCP from ..backend_client import BackendRequestContext from ..config import DEFAULT_PROFILE from ..errors import QingflowApiError, backend_code_int, backend_code_value_int, is_auth_like_error, message_looks_like_invalid_token, raise_tool_error from ..id_utils import ids_equal, normalize_positive_id_int, normalize_positive_id_text, stringify_backend_id from ..json_types import JSONObject from .approval_tools import ApprovalTools, _approval_page_amount, _approval_page_items, _approval_page_total from .base import ToolBase from .qingbi_report_tools import _qingbi_base_url, _should_retry_asos_data from .record_tools import ( FieldIndex, LAYOUT_ONLY_QUE_TYPES, SUBTABLE_QUE_TYPES, RecordTools, _build_answer_backed_field_index, _build_applicant_hidden_linked_top_level_field_index, _build_applicant_top_level_field_index, _build_static_schema_linkage_payloads, _canonical_value_is_empty, _canonicalize_answer_value_for_compare, _clone_form_field, _coerce_count, _collect_linked_required_field_ids, _collect_option_linked_field_ids, _collect_question_relations, _field_ref_payload, _merge_field_indexes, _normalize_form_schema, _subtable_descendant_ids, ) from .task_tools import TaskTools, _task_page_amount, _task_page_items, _task_page_total TASK_LOCATOR_MAX_SCAN_PAGES_PER_BOX = 3 class TaskContextTools(ToolBase): """任务上下文工具（中文名：任务上下文与审批执行）。 类型：任务深度上下文工具。 主要职责： 1. 聚合任务详情、候选人、关联报表与流程日志； 2. 执行审批动作（通过、驳回、转交等）； 3. 为任务处理过程提供可执行上下文而非仅列表数据。 """ def __init__(self, sessions, backend) -> None: # type: ignore[no-untyped-def] """执行内部辅助逻辑。""" super().__init__(sessions, backend) self._task_tools = TaskTools(sessions, backend) self._approval_tools = ApprovalTools(sessions, backend) self._record_tools = RecordTools(sessions, backend) self._app_name_cache: dict[str, str | None] = {} def register(self, mcp: FastMCP) -> None: """注册当前工具到 MCP 服务。""" @mcp.tool( description=( "List workflow tasks. `query` first uses backend task search; if the backend returns zero rows, " "public task_list falls back to local matching on app_name, workflow_node_name, app_key, and record_id." ) ) def task_list( profile: str = DEFAULT_PROFILE, task_box: str = "todo", flow_status: str = "all", app_key: str | None = None, workflow_node_id: int | None = None, query: str | None = None, page: int = 1, page_size: int = 20, ) -> dict[str, Any]: return self.task_list( profile=profile, task_box=task_box, flow_status=flow_status, app_key=app_key, workflow_node_id=workflow_node_id, query=query, page=page, page_size=page_size, ) @mcp.tool( description=( "Read one workflow task. Prefer task_id from task_list.data.items[].task_id; " "task_id is not a row number, list index, record id, or workflow node id." ) ) def task_get( profile: str = DEFAULT_PROFILE, task_id: str = "", include_candidates: bool = True, include_associated_reports: bool = True, ) -> dict[str, Any]: return self.task_get( profile=profile, task_id=task_id, app_key="", record_id="", workflow_node_id=0, include_candidates=include_candidates, include_associated_reports=include_associated_reports, ) @mcp.tool( description=( self._high_risk_tool_description(operation="execute", target="workflow task action") + " Pass task_id from task_list.data.items[].task_id. Do not pass a row number, list index, record id, or workflow node id as task_id." ) ) def task_action_execute( profile: str = DEFAULT_PROFILE, task_id: str = "", action: str = "", payload: dict[str, Any] | None = None, fields: dict[str, Any] | None = None, ) -> dict[str, Any]: return self.task_action_execute( profile=profile, task_id=task_id, action=action, payload=payload or {}, fields=fields or {}, ) @mcp.tool( description=( "Read a task-associated report. Pass task_id from task_list.data.items[].task_id; " "task_id is not a row number, list index, record id, or workflow node id." ) ) def task_associated_report_detail_get( profile: str = DEFAULT_PROFILE, task_id: str = "", report_id: int = 0, page: int = 1, page_size: int = 20, ) -> dict[str, Any]: return self.task_associated_report_detail_get( profile=profile, task_id=task_id, app_key="", record_id="", workflow_node_id=0, report_id=report_id, page=page, page_size=page_size, ) @mcp.tool( description=( "Read workflow log for one task context. Pass task_id from task_list.data.items[].task_id; " "task_id is not a row number, list index, record id, or workflow node id." ) ) def task_workflow_log_get( profile: str = DEFAULT_PROFILE, task_id: str = "", ) -> dict[str, Any]: return self.task_workflow_log_get( profile=profile, task_id=task_id, app_key="", record_id="", workflow_node_id=0, ) def task_list( self, *, profile: str, task_box: str, flow_status: str, app_key: str | None, workflow_node_id: int | None, query: str | None, page: int, page_size: int, ) -> dict[str, Any]: """执行任务相关逻辑。""" response = self._list_normalized_task_items( profile=profile, task_box=task_box, flow_status=flow_status, app_key=app_key, workflow_node_id=workflow_node_id, query=query, page=page, page_size=page_size, ) warnings: list[dict[str, Any]] = [] items = response["items"] if isinstance(response.get("items"), list) else [] page_amount = response.get("page_amount") reported_total = response.get("reported_total") if query and not items: fallback = self._task_list_local_query_fallback( profile=profile, task_box=task_box, flow_status=flow_status, app_key=app_key, workflow_node_id=workflow_node_id, query=query, page=page, page_size=page_size, ) if fallback is not None: items = fallback["items"] returned_items = len(items) page_amount = fallback["page_amount"] reported_total = fallback["reported_total"] warnings.append( { "code": "TASK_LIST_QUERY_FALLBACK_APPLIED", "message": ( "backend searchKey returned zero tasks; task_list fell back to local matching on " "app_name, workflow_node_name, app_key, and record_id." ), } ) public_items = [self._public_task_item(item) for item in items] return { "profile": profile, "ws_id": response.get("raw", {}).get("ws_id") if isinstance(response.get("raw"), dict) else None, "ok": True, "request_route": response.get("raw", {}).get("request_route") if isinstance(response.get("raw"), dict) else None, "warnings": warnings, "output_profile": "normal", "data": { "items": public_items, "pagination": { "page": page, "page_size": page_size, "returned_items": len(public_items), "page_amount": page_amount, "reported_total": reported_total, }, "selection": { "app_key": app_key, "workflow_node_id": workflow_node_id, "query": query, }, }, } def task_get( self, *, profile: str, task_id: Any = None, app_key: str = "", record_id: Any = "", workflow_node_id: int = 0, include_candidates: bool = True, include_associated_reports: bool = True, ) -> dict[str, Any]: """执行任务相关逻辑。""" if task_id in (None, ""): normalize_positive_id_int(record_id, field_name="record_id") def runner(session_profile, context): locator = self._resolve_task_locator_input( profile=profile, task_id=task_id, app_key=app_key, record_id=record_id, workflow_node_id=workflow_node_id, ) task_id_text = locator["task_id"] resolved_app_key = str(locator["app_key"]) resolved_record_id = int(locator["record_id"]) resolved_workflow_node_id = int(locator["workflow_node_id"]) resolved_task_box = str(locator.get("task_box") or "todo") self._require_app_record_and_node(resolved_app_key, resolved_record_id, resolved_workflow_node_id) data = self._build_task_context( profile=profile, context=context, app_key=resolved_app_key, record_id=resolved_record_id, workflow_node_id=resolved_workflow_node_id, task_box=resolved_task_box, include_candidates=include_candidates, include_associated_reports=include_associated_reports, current_uid=session_profile.uid, ) context_warnings = data.get("warnings") if isinstance(data.get("warnings"), list) else [] data = self._compact_task_get_context(data) task_payload = data.get("task") if isinstance(task_payload, dict) and task_id_text is not None: task_payload["task_id"] = task_id_text return { "profile": profile, "ws_id": session_profile.selected_ws_id, "ok": True, "request_route": self._request_route_payload(context), "warnings": context_warnings, "output_profile": "normal", "data": data, } return self._run(profile, runner, tool_name='任务上下文详情') def task_save_only( self, *, profile: str, task_id: Any, fields: dict[str, Any] | None = None, ) -> dict[str, Any]: """执行任务相关逻辑。""" field_updates = dict(fields or {}) if not field_updates: raise_tool_error(QingflowApiError.config_error("fields is required and must be non-empty for task_save_only")) return self.task_action_execute( profile=profile, task_id=task_id, action="save_only", payload={}, fields=field_updates, ) def task_action_execute( self, *, profile: str, task_id: Any, action: str, payload: dict[str, Any], fields: dict[str, Any] | None = None, ) -> dict[str, Any]: """执行任务相关逻辑。""" if task_id in (None, ""): raise_tool_error( QingflowApiError.config_error( "task_id is required for task_action_execute; get it from task_list.data.items[].task_id" ) ) return self._task_action_execute_with_locator( profile=profile, task_id=task_id, app_key="", record_id="", workflow_node_id=0, action=action, payload=payload, fields=fields, ) def _task_action_execute_with_locator( self, *, profile: str, task_id: Any = None, app_key: str = "", record_id: Any = "", workflow_node_id: int = 0, action: str, payload: dict[str, Any], fields: dict[str, Any] | None = None, ) -> dict[str, Any]: if task_id in (None, ""): normalize_positive_id_int(record_id, field_name="record_id") normalized_action = (action or "").strip().lower() if normalized_action not in {"approve", "reject", "rollback", "transfer", "urge", "save_only"}: raise_tool_error( QingflowApiError.not_supported( "TASK_ACTION_UNSUPPORTED: action must be one of approve, reject, rollback, transfer, urge, or save_only" ) ) body = dict(payload or {}) field_updates = dict(fields or {}) if field_updates and body.get("answers") is not None: raise_tool_error( QingflowApiError.config_error( "task actions must not provide payload.answers and fields at the same time; pass field changes through fields only" ) ) def runner(session_profile, context): locator = self._resolve_task_locator_input( profile=profile, task_id=task_id, app_key=app_key, record_id=record_id, workflow_node_id=workflow_node_id, ) task_id_text = locator["task_id"] resolved_app_key = str(locator["app_key"]) resolved_record_id = int(locator["record_id"]) resolved_record_id_text = str(locator["record_id_text"] or "") resolved_workflow_node_id = int(locator["workflow_node_id"]) resolved_task_box = str(locator.get("task_box") or "todo") record_id_text = resolved_record_id_text self._require_app_record_and_node(resolved_app_key, resolved_record_id, resolved_workflow_node_id) if normalized_action == "urge": raw = self._execute_task_action( profile=profile, app_key=resolved_app_key, record_id=resolved_record_id, workflow_node_id=resolved_workflow_node_id, normalized_action=normalized_action, payload=body, prepared_fields=None, ) verification, verification_warnings = self._verify_task_action_runtime( profile=profile, context=context, app_key=resolved_app_key, record_id=resolved_record_id, workflow_node_id=resolved_workflow_node_id, action=normalized_action, before_apply_status=None, runtime_baseline=None, ) result = { "profile": raw.get("profile", profile), "ws_id": raw.get("ws_id", session_profile.selected_ws_id), "ok": bool(raw.get("ok", True)), "status": "success", "error_code": None, "action_executed": True, "safe_to_retry": False, "request_route": raw.get("request_route") or self._request_route_payload(context), "warnings": verification_warnings, "verification": verification, "output_profile": "normal", "data": { "action": normalized_action, "resource": { "app_key": resolved_app_key, "record_id": record_id_text, "workflow_node_id": resolved_workflow_node_id, }, "selection": {"action": normalized_action}, "result": raw.get("result"), "human_review": True, "field_update_applied": False, }, } if task_id_text is not None: result["data"]["resource"]["task_id"] = task_id_text return result if resolved_task_box != "todo": raise_tool_error( QingflowApiError.config_error( f"task_id={task_id_text or ''} resolved to task_box='{resolved_task_box}', but task_action_execute can only execute current todo tasks" ) ) try: task_context = self._build_task_context( profile=profile, context=context, app_key=resolved_app_key, record_id=resolved_record_id, workflow_node_id=resolved_workflow_node_id, task_box=resolved_task_box, include_candidates=False, include_associated_reports=False, current_uid=session_profile.uid, ) except QingflowApiError as error: if backend_code_int(error) == 46001: return self._task_action_visibility_unverified_response( profile=profile, session_profile=session_profile, context=context, app_key=resolved_app_key, record_id=resolved_record_id, workflow_node_id=resolved_workflow_node_id, action=normalized_action, source_error=error, before_apply_status=None, task_id=task_id_text, ) raise if normalized_action == "save_only" and not field_updates: raise_tool_error( QingflowApiError.config_error("fields is required and must be non-empty for action 'save_only'") ) if normalized_action == "transfer": target_member_id = self._extract_positive_int(body, "target_member_id", aliases=("uid", "targetMemberId")) if target_member_id == session_profile.uid: raise_tool_error( QingflowApiError.config_error( "task transfer does not support transferring to the current user; choose another transfer member" ) ) capabilities = task_context.get("capabilities") or {} available_actions = capabilities.get("available_actions") or [] if normalized_action not in available_actions: if normalized_action == "save_only": capability_warnings = capabilities.get("warnings") or [] message = ( "task action 'save_only' is not currently available for the current node; " "MCP only exposes save_only when backend editableQueIds returns a non-empty result" ) if capability_warnings: message += "; backend editableQueIds is unavailable or empty for this task context" raise_tool_error(QingflowApiError.config_error(message)) raise_tool_error( QingflowApiError.config_error( f"task action '{normalized_action}' is not currently available for task_id='{task_id_text}'" ) ) feedback_required_for = capabilities.get("action_constraints", {}).get("feedback_required_for") or [] if normalized_action in feedback_required_for and not self._extract_audit_feedback(body): raise_tool_error( QingflowApiError.config_error( f"payload.audit_feedback is required for action '{normalized_action}' on the current node" ) ) if normalized_action == "urge" and field_updates: raise_tool_error( QingflowApiError.not_supported( "TASK_ACTION_FIELDS_NOT_SUPPORTED: action 'urge' does not support fields because the downstream route does not accept task answers" ) ) prepared_fields = None if field_updates: prepared_fields = self._prepare_task_field_update( profile=profile, context=context, app_key=resolved_app_key, record_id=resolved_record_id, workflow_node_id=resolved_workflow_node_id, task_context=task_context, fields=field_updates, ) before_apply_status = ((task_context.get("record") or {}).get("apply_status")) runtime_baseline = None if normalized_action != "save_only": runtime_baseline = self._capture_task_runtime_baseline( profile=profile, context=context, app_key=resolved_app_key, record_id=resolved_record_id, workflow_node_id=resolved_workflow_node_id, ) try: raw = self._execute_task_action( profile=profile, app_key=resolved_app_key, record_id=resolved_record_id, workflow_node_id=resolved_workflow_node_id, normalized_action=normalized_action, payload=body, prepared_fields=prepared_fields, ) except QingflowApiError as error: if backend_code_int(error) == 46001: return self._task_action_visibility_unverified_response( profile=profile, session_profile=session_profile, context=context, app_key=resolved_app_key, record_id=resolved_record_id, workflow_node_id=resolved_workflow_node_id, action=normalized_action, source_error=error, before_apply_status=before_apply_status, task_id=task_id_text, ) raise if normalized_action == "save_only": verification, warnings = self._verify_task_save_only( context=context, app_key=resolved_app_key, record_id=resolved_record_id, workflow_node_id=resolved_workflow_node_id, before_apply_status=before_apply_status, expected_answers=((prepared_fields or {}).get("normalized_answers") or []), task_context=task_context, ) save_verified = bool(verification.get("fields_saved_verified")) and bool(verification.get("task_still_actionable")) status = "success" if save_verified else "failed" error_code = None if save_verified else "TASK_SAVE_ONLY_VERIFICATION_FAILED" else: verification, warnings = self._verify_task_action_runtime( profile=profile, context=context, app_key=resolved_app_key, record_id=resolved_record_id, workflow_node_id=resolved_workflow_node_id, action=normalized_action, before_apply_status=before_apply_status, runtime_baseline=runtime_baseline, ) runtime_verified = bool(verification.get("runtime_continuation_verified")) status = "success" if runtime_verified else "partial_success" error_code = None if runtime_verified else "WORKFLOW_CONTINUATION_UNVERIFIED" result = { "profile": raw.get("profile", profile), "ws_id": raw.get("ws_id", session_profile.selected_ws_id), "ok": bool(raw.get("ok", True)) and status != "failed", "status": status, "error_code": error_code, "action_executed": True, "safe_to_retry": False, "request_route": raw.get("request_route") or self._request_route_payload(context), "warnings": warnings, "verification": verification, "output_profile": "normal", "data": { "action": normalized_action, "resource": { "app_key": resolved_app_key, "record_id": record_id_text, "workflow_node_id": resolved_workflow_node_id, }, "selection": {"action": normalized_action}, "result": raw.get("result"), "human_review": True, "field_update_applied": bool(field_updates), }, } if task_id_text is not None: resource = result["data"].get("resource") if isinstance(resource, dict): resource["task_id"] = task_id_text return result return self._run(profile, runner, tool_name="执行任务动作") def _execute_task_action( self, *, profile: str, app_key: str, record_id: Any, workflow_node_id: int, normalized_action: str, payload: dict[str, Any], prepared_fields: dict[str, Any] | None, ) -> dict[str, Any]: """执行内部辅助逻辑。""" merged_answers = None normalized_answers = None if isinstance(prepared_fields, dict): candidate_answers = prepared_fields.get("merged_answers") if isinstance(candidate_answers, list): merged_answers = candidate_answers candidate_normalized = prepared_fields.get("normalized_answers") if isinstance(candidate_normalized, list): normalized_answers = candidate_normalized if normalized_action == "approve": action_payload = dict(payload) action_payload["nodeId"] = workflow_node_id if merged_answers is not None: action_payload["answers"] = merged_answers return self._approval_tools.record_approve( profile=profile, app_key=app_key, apply_id=record_id, payload=action_payload, ) if normalized_action == "reject": action_payload = dict(payload) action_payload["nodeId"] = workflow_node_id if merged_answers is not None: action_payload["answers"] = merged_answers if not self._extract_audit_feedback(action_payload): raise_tool_error(QingflowApiError.config_error("payload.audit_feedback is required for reject")) return self._approval_tools.record_reject( profile=profile, app_key=app_key, apply_id=record_id, payload=action_payload, ) if normalized_action == "rollback": target_node_id = self._extract_positive_int(payload, "target_workflow_node_id", aliases=("targetAuditNodeId", "targetWorkflowNodeId")) action_payload: JSONObject = { "auditNodeId": workflow_node_id, "targetAuditNodeId": target_node_id, } audit_feedback = self._extract_audit_feedback(payload) if audit_feedback: action_payload["auditFeedback"] = audit_feedback if merged_answers is not None: action_payload["answers"] = merged_answers return self._approval_tools.record_rollback( profile=profile, app_key=app_key, apply_id=record_id, payload=action_payload, ) if normalized_action == "transfer": target_member_id = self._extract_positive_int(payload, "target_member_id", aliases=("uid", "targetMemberId")) action_payload = { "auditNodeId": workflow_node_id, "uid": target_member_id, } audit_feedback = self._extract_audit_feedback(payload) if audit_feedback: action_payload["auditFeedback"] = audit_feedback if merged_answers is not None: action_payload["answers"] = merged_answers return self._approval_tools.record_transfer( profile=profile, app_key=app_key, apply_id=record_id, payload=action_payload, ) if normalized_action == "save_only": if normalized_answers is None: raise_tool_error(QingflowApiError.config_error("fields is required for action 'save_only'")) return self._task_save_only( profile=profile, app_key=app_key, record_id=record_id, workflow_node_id=workflow_node_id, apply_answers=normalized_answers, ) return self._task_tools.task_urge( profile=profile, app_key=app_key, row_record_id=record_id, ) def _verify_task_action_runtime( self, *, profile: str, context: BackendRequestContext, app_key: str, record_id: int, workflow_node_id: int, action: str, before_apply_status: Any, runtime_baseline: dict[str, Any] | None = None, ) -> tuple[dict[str, Any], list[dict[str, Any]]]: """执行内部辅助逻辑。""" verification: dict[str, Any] = { "action_executed": True, "runtime_continuation_verified": action == "urge", "scope": "workflow_runtime", "task_context_visibility_verified": True, } warnings: list[dict[str, Any]] = [] if action == "urge": return verification, warnings state_after: dict[str, Any] | None = None try: state_after = self.backend.request( "GET", context, f"/app/{app_key}/apply/{record_id}", params={"role": 3, "listType": 1, "auditNodeId": workflow_node_id}, ) verification["record_state_readable"] = True verification["before_apply_status"] = before_apply_status verification["after_apply_status"] = state_after.get("applyStatus") if isinstance(state_after, dict) else None verification["record_state_changed"] = verification["after_apply_status"] != before_apply_status except QingflowApiError as error: verification["record_state_readable"] = False verification["record_state_changed"] = False verification["record_state_error"] = { "http_status": error.http_status, "backend_code": error.backend_code, "category": error.category, "request_id": error.request_id, "message": error.message, } log_items: list[dict[str, Any]] = [] try: log_page = self.backend.request( "POST", context, "/application/workflow/node/record", json_body={ "key": app_key, "rowRecordId": record_id, "nodeId": workflow_node_id, "role": 3, "pageNum": 1, "pageSize": 50, }, ) log_items = self._normalize_workflow_logs(log_page) verification["workflow_log_visible"] = True verification["workflow_log_count"] = len(log_items) except QingflowApiError as error: verification["workflow_log_visible"] = False verification["workflow_log_count"] = None verification["workflow_log_error"] = { "http_status": error.http_status, "backend_code": error.backend_code, "category": error.category, "request_id": error.request_id, "message": error.message, } todo_items = self._safe_task_list_items(profile=profile, task_box="todo", app_key=app_key) initiated_items = self._safe_task_list_items(profile=profile, task_box="initiated", app_key=app_key) downstream_todo_detected = any( ids_equal(item.get("record_id"), record_id) and int(item.get("workflow_node_id") or 0) != workflow_node_id for item in todo_items if isinstance(item, dict) ) initiated_visible = any( ids_equal(item.get("record_id"), record_id) for item in initiated_items if isinstance(item, dict) ) verification["downstream_todo_detected"] = downstream_todo_detected verification["initiated_task_visible"] = initiated_visible baseline_downstream_nodes = set() baseline_log_count = None baseline_log_digest = None if isinstance(runtime_baseline, dict): baseline_downstream_nodes = set(runtime_baseline.get("downstream_todo_nodes") or []) baseline_log_count = runtime_baseline.get("workflow_log_count") baseline_log_digest = runtime_baseline.get("workflow_log_digest") current_downstream_nodes = { int(item.get("workflow_node_id") or 0) for item in todo_items if isinstance(item, dict) and ids_equal(item.get("record_id"), record_id) and int(item.get("workflow_node_id") or 0) != workflow_node_id } workflow_log_digest = self._workflow_log_digest(log_items) verification["downstream_todo_nodes"] = sorted(node_id for node_id in current_downstream_nodes if node_id > 0) verification["downstream_todo_changed"] = current_downstream_nodes != baseline_downstream_nodes verification["workflow_log_advanced"] = bool( verification.get("workflow_log_visible") and ( (isinstance(baseline_log_count, int) and len(log_items) > baseline_log_count) or (baseline_log_digest is not None and workflow_log_digest is not None and workflow_log_digest != baseline_log_digest) ) ) runtime_verified = bool( verification.get("record_state_changed") or verification.get("downstream_todo_changed") or verification.get("workflow_log_advanced") ) record_state_error = verification.get("record_state_error") runtime_consumed_after_action = bool( runtime_verified and isinstance(record_state_error, dict) and backend_code_value_int(record_state_error.get("backend_code")) == 46001 ) if runtime_consumed_after_action: verification["record_state_scope"] = "current_node_runtime" verification["record_state_unavailable_reason"] = "runtime_consumed_after_action" verification["record_state_unavailability_expected"] = True warnings.append( { "code": "TASK_RUNTIME_CONSUMED_AFTER_ACTION", "message": ( "the current workflow node runtime is no longer readable after the action (backend 46001), " "which usually means the node has been consumed and the workflow has already continued." ), } ) permission_blocked_sources: list[str] = [] if ( isinstance(verification.get("record_state_error"), dict) and _is_permission_context_error_payload(verification["record_state_error"]) ): permission_blocked_sources.append("record_state") if ( isinstance(verification.get("workflow_log_error"), dict) and _is_permission_context_error_payload(verification["workflow_log_error"]) ): permission_blocked_sources.append("workflow_log") if permission_blocked_sources: warnings.append( { "code": "TASK_ACTION_VERIFICATION_PERMISSION_UNAVAILABLE", "message": ( "task action executed, but some post-action verification reads are unavailable " "in this permission context; do not treat the verification read failure as action denial." ), "sources": permission_blocked_sources, } ) verification["runtime_continuation_verified"] = runtime_verified if not runtime_verified: warnings.append( { "code": "WORKFLOW_CONTINUATION_UNVERIFIED", "message": "task action executed, but MCP could not verify downstream workflow continuation from record state, workflow logs, or downstream todo tasks.", } ) return verification, warnings def _verify_task_save_only( self, *, context: BackendRequestContext, app_key: str, record_id: int, workflow_node_id: int, before_apply_status: Any, expected_answers: list[dict[str, Any]], task_context: dict[str, Any], ) -> tuple[dict[str, Any], list[dict[str, Any]]]: """执行内部辅助逻辑。""" verification: dict[str, Any] = { "action_executed": True, "scope": "task_field_save", "runtime_continuation_verified": False, "task_context_visibility_verified": True, "fields_saved_verified": False, "task_still_actionable": False, "workflow_not_advanced": False, "before_apply_status": before_apply_status, } warnings: list[dict[str, Any]] = [] try: detail = self.backend.request( "GET", context, f"/app/{app_key}/apply/{record_id}", params={"role": 3, "listType": 1, "auditNodeId": workflow_node_id}, ) except QingflowApiError as error: verification["record_state_readable"] = False verification["task_context_visibility_verified"] = False verification["transport_error"] = { "http_status": error.http_status, "backend_code": error.backend_code, "category": error.category, } warnings.append( { "code": "TASK_SAVE_ONLY_UNVERIFIED", "message": "save_only write was sent, but MCP could not re-read the current task context to verify that the node remains actionable and fields were saved.", } ) return verification, warnings verification["record_state_readable"] = True verification["task_still_actionable"] = True after_apply_status = detail.get("applyStatus") if isinstance(detail, dict) else None verification["after_apply_status"] = after_apply_status verification["workflow_not_advanced"] = after_apply_status == before_apply_status current_record = task_context.get("record") if isinstance(task_context.get("record"), dict) else {} actual_answers = detail.get("answers") if isinstance(detail, dict) and isinstance(detail.get("answers"), list) else [] expected_by_id = { que_id: answer for answer in expected_answers if isinstance(answer, dict) and (que_id := _coerce_count(answer.get("queId"))) is not None and que_id > 0 } actual_by_id = { que_id: answer for answer in actual_answers if isinstance(answer, dict) and (que_id := _coerce_count(answer.get("queId"))) is not None and que_id > 0 } update_schema = task_context.get("update_schema") if isinstance(task_context.get("update_schema"), dict) else {} writable_titles = { item.get("title"): item for item in (update_schema.get("writable_fields") or []) if isinstance(item, dict) and item.get("title") } missing_fields: list[dict[str, Any]] = [] mismatched_fields: list[dict[str, Any]] = [] for que_id, expected in expected_by_id.items(): actual = actual_by_id.get(que_id) title = None if isinstance(current_record, dict): for answer in (current_record.get("answers") or []): if isinstance(answer, dict) and _coerce_count(answer.get("queId")) == que_id: title = answer.get("queTitle") break if title is None: title = next((key for key, item in writable_titles.items() if isinstance(item, dict) and item.get("field_id") == que_id), None) field_payload = {"que_id": que_id, "que_title": title} if actual is None: missing_fields.append(field_payload) continue expected_value = _canonicalize_answer_value_for_compare(expected, None) actual_value = _canonicalize_answer_value_for_compare(actual, None) if _canonical_value_is_empty(expected_value): continue if actual_value != expected_value: mismatched_fields.append( { **field_payload, "expected": expected_value, "actual": actual_value, } ) verification["missing_fields"] = missing_fields verification["mismatched_fields"] = mismatched_fields verification["fields_saved_verified"] = not missing_fields and not mismatched_fields if not verification["workflow_not_advanced"]: warnings.append( { "code": "TASK_SAVE_ONLY_ADVANCED_WORKFLOW", "message": "save_only unexpectedly changed the workflow runtime state; the task should have remained on the current node.", } ) if not verification["fields_saved_verified"]: warnings.append( { "code": "TASK_SAVE_ONLY_FIELD_VERIFICATION_FAILED", "message": "save_only completed, but MCP could not verify that all requested field changes were persisted on the current task node.", } ) return verification, warnings def _capture_task_runtime_baseline( self, *, profile: str, context: BackendRequestContext, app_key: str, record_id: int, workflow_node_id: int, ) -> dict[str, Any]: """执行内部辅助逻辑。""" baseline: dict[str, Any] = { "workflow_log_visible": False, "workflow_log_count": None, "workflow_log_digest": None, "downstream_todo_nodes": [], } try: log_page = self.backend.request( "POST", context, "/application/workflow/node/record", json_body={ "key": app_key, "rowRecordId": record_id, "nodeId": workflow_node_id, "role": 3, "pageNum": 1, "pageSize": 50, }, ) log_items = self._normalize_workflow_logs(log_page) baseline["workflow_log_visible"] = True baseline["workflow_log_count"] = len(log_items) baseline["workflow_log_digest"] = self._workflow_log_digest(log_items) except QingflowApiError as exc: if is_auth_like_error(exc): raise pass todo_items = self._safe_task_list_items(profile=profile, task_box="todo", app_key=app_key) baseline["downstream_todo_nodes"] = sorted( { int(item.get("workflow_node_id") or 0) for item in todo_items if isinstance(item, dict) and ids_equal(item.get("record_id"), record_id) and int(item.get("workflow_node_id") or 0) != workflow_node_id } ) return baseline def _task_action_visibility_unverified_response( self, *, profile: str, session_profile, context: BackendRequestContext, app_key: str, record_id: int, workflow_node_id: int, action: str, source_error: QingflowApiError, before_apply_status: Any, task_id: str | None = None, ) -> dict[str, Any]: """执行内部辅助逻辑。""" record_id_text = stringify_backend_id(record_id) verification, warnings = self._verify_task_action_runtime( profile=profile, context=context, app_key=app_key, record_id=record_id, workflow_node_id=workflow_node_id, action=action, before_apply_status=before_apply_status, ) verification["action_executed"] = False verification["task_context_visibility_verified"] = bool(verification.get("runtime_continuation_verified")) if verification["task_context_visibility_verified"]: warnings.append( { "code": "TASK_ALREADY_PROCESSED_UNCONFIRMED_ACTOR", "message": "the task is no longer actionable in the current context; MCP found downstream workflow evidence and treats it as already processed by another actor.", } ) result = { "profile": profile, "ws_id": session_profile.selected_ws_id, "ok": True, "status": "partial_success", "error_code": "TASK_ALREADY_PROCESSED", "request_route": self._request_route_payload(context), "warnings": warnings, "verification": verification, "output_profile": "normal", "data": { "action": action, "resource": { "app_key": app_key, "record_id": record_id_text, "workflow_node_id": workflow_node_id, }, "selection": {"action": action}, "result": None, "human_review": True, }, } if task_id is not None: resource = result["data"].get("resource") if isinstance(resource, dict): resource["task_id"] = task_id return result warnings.append( { "code": "TASK_CONTEXT_VISIBILITY_UNVERIFIED", "message": "the task is no longer actionable, and MCP could not verify from state or workflow logs whether it was already processed.", } ) result = { "profile": profile, "ws_id": session_profile.selected_ws_id, "ok": False, "status": "failed", "error_code": "TASK_CONTEXT_VISIBILITY_UNVERIFIED", "request_route": self._request_route_payload(context), "warnings": warnings, "verification": verification, "output_profile": "normal", "data": { "action": action, "resource": { "app_key": app_key, "record_id": record_id_text, "workflow_node_id": workflow_node_id, }, "selection": {"action": action}, "result": None, "human_review": True, "transport_error": { "http_status": source_error.http_status, "backend_code": source_error.backend_code, "category": source_error.category, }, }, } if task_id is not None: resource = result["data"].get("resource") if isinstance(resource, dict): resource["task_id"] = task_id return result def _safe_task_list_items(self, *, profile: str, task_box: str, app_key: str) -> list[dict[str, Any]]: """执行内部辅助逻辑。""" try: response = self._list_normalized_task_items( profile=profile, task_box=task_box, flow_status="all", app_key=app_key, workflow_node_id=None, query=None, page=1, page_size=50, ) except QingflowApiError as exc: if not _is_task_optional_read_error(exc): raise return [] items = response.get("items") if isinstance(response, dict) else None if not isinstance(items, list): return [] return [item for item in items if isinstance(item, dict)] def _list_normalized_task_items( self, *, profile: str, task_box: str, flow_status: str, app_key: str | None, workflow_node_id: int | None, query: str | None, page: int, page_size: int, ) -> dict[str, Any]: normalized_type = self._task_tools._task_box_to_type(task_box) normalized_status = self._task_tools._flow_status_to_process_status(flow_status) raw = self._task_tools.task_list( profile=profile, type=normalized_type, process_status=normalized_status, app_key=app_key, node_id=workflow_node_id, search_key=query, page_num=page, page_size=page_size, create_time_asc=None, ) task_page = raw.get("page", {}) return { "raw": raw, "items": [self._normalize_task_item(item) for item in _task_page_items(task_page) if isinstance(item, dict)], "page_amount": _task_page_amount(task_page), "reported_total": _task_page_total(task_page), } def _task_list_local_query_fallback( self, *, profile: str, task_box: str, flow_status: str, app_key: str | None, workflow_node_id: int | None, query: str, page: int, page_size: int, ) -> dict[str, Any] | None: scan_page_size = max(page_size, 100) scan_page = 1 page_amount: int | None = None matched_items: list[dict[str, Any]] = [] while True: response = self._list_normalized_task_items( profile=profile, task_box=task_box, flow_status=flow_status, app_key=app_key, workflow_node_id=workflow_node_id, query=None, page=scan_page, page_size=scan_page_size, ) normalized_items = response.get("items") if isinstance(response.get("items"), list) else [] matched_items.extend(item for item in normalized_items if self._task_item_matches_query(item, query)) if page_amount is None: coerced_page_amount = _coerce_count(response.get("page_amount")) if coerced_page_amount is not None and coerced_page_amount > 0: page_amount = coerced_page_amount if page_amount is not None and scan_page >= page_amount: break if not normalized_items or len(normalized_items) < scan_page_size: break scan_page += 1 if not matched_items: return None start = max(page - 1, 0) * page_size end = start + page_size matched_total = len(matched_items) matched_page_amount = (matched_total + page_size - 1) // page_size if page_size > 0 else 0 return { "items": matched_items[start:end], "page_amount": matched_page_amount, "reported_total": matched_total, } def _resolve_task_locator_by_task_id(self, *, profile: str, task_id: Any) -> dict[str, Any]: task_id_text = normalize_positive_id_text(task_id, field_name="task_id") searched_task_boxes = ("todo", "initiated", "cc", "done") incomplete_task_boxes: list[str] = [] inaccessible_task_boxes: list[dict[str, Any]] = [] truncated_task_boxes: list[dict[str, Any]] = [] page_size = 100 for task_box in searched_task_boxes: page = 1 page_amount: int | None = None while True: try: response = self._list_normalized_task_items( profile=profile, task_box=task_box, flow_status="all", app_key=None, workflow_node_id=None, query=None, page=page, page_size=page_size, ) except QingflowApiError as exc: if not _is_optional_task_box_locator_error(exc): raise inaccessible_task_boxes.append( { "task_box": task_box, "backend_code": exc.backend_code, "http_status": exc.http_status, "request_id": exc.request_id, } ) break items = response.get("items") if isinstance(response.get("items"), list) else [] for item in items: if not isinstance(item, dict) or not ids_equal(item.get("task_id"), task_id_text): continue app_key = str(item.get("app_key") or "").strip() record_id = stringify_backend_id(item.get("record_id")) workflow_node_id = int(item.get("workflow_node_id") or 0) if not app_key or record_id is None or workflow_node_id <= 0: incomplete_task_boxes.append(task_box) continue return { "task_id": task_id_text, "task_box": task_box, "app_key": app_key, "record_id": record_id, "workflow_node_id": workflow_node_id, } if page_amount is None: coerced_page_amount = _coerce_count(response.get("page_amount")) if coerced_page_amount is not None and coerced_page_amount > 0: page_amount = coerced_page_amount if page_amount is not None and page >= page_amount: break if not items or len(items) < page_size: break if page >= TASK_LOCATOR_MAX_SCAN_PAGES_PER_BOX: truncated_task_boxes.append( { "task_box": task_box, "max_pages": TASK_LOCATOR_MAX_SCAN_PAGES_PER_BOX, "page_size": page_size, "reported_page_amount": page_amount, } ) break page += 1 if incomplete_task_boxes: searched = ", ".join(incomplete_task_boxes) raise_tool_error( QingflowApiError.config_error( f"task_id={task_id_text} resolved to an incomplete task locator in task_box={searched}; rerun task_list and pass the exact data.items[].task_id. Do not substitute a row number or rebuild the locator from app_key/record_id/workflow_node_id." ) ) if inaccessible_task_boxes: searched = ", ".join(str(item.get("task_box")) for item in inaccessible_task_boxes) raise_tool_error( QingflowApiError.config_error( f"task_id={task_id_text} was not found in visible task boxes; some task boxes were not searchable in the current permission context: {searched}. Rerun task_list and use data.items[].task_id; do not substitute a row number or rebuild the locator.", details={ "task_id": task_id_text, "searched_task_boxes": list(searched_task_boxes), "inaccessible_task_boxes": inaccessible_task_boxes, }, ) ) if truncated_task_boxes: raise_tool_error( QingflowApiError.config_error( f"task_id={task_id_text} was not found in the first {TASK_LOCATOR_MAX_SCAN_PAGES_PER_BOX} pages of each visible task box. Rerun task_list with a query or the relevant page and pass the exact data.items[].task_id; do not guess task ids.", details={ "task_id": task_id_text, "searched_task_boxes": list(searched_task_boxes), "truncated_task_boxes": truncated_task_boxes, "max_pages_per_task_box": TASK_LOCATOR_MAX_SCAN_PAGES_PER_BOX, "page_size": page_size, }, ) ) raise_tool_error( QingflowApiError.config_error( f"task_id={task_id_text} was not found in the current visible task boxes (todo, initiated, cc, done). Rerun task_list and pass data.items[].task_id; do not use the displayed row number or guess app_key/record_id/workflow_node_id." ) ) def _resolve_task_locator_input( self, *, profile: str, task_id: Any = None, app_key: str = "", record_id: Any = "", workflow_node_id: int = 0, ) -> dict[str, Any]: task_id_text = normalize_positive_id_text(task_id, field_name="task_id") if task_id not in (None, "") else None resolved_app_key = (app_key or "").strip() resolved_record_id: int resolved_workflow_node_id: int if task_id_text is not None: locator = self._resolve_task_locator_by_task_id(profile=profile, task_id=task_id_text) resolved_app_key = str(locator["app_key"]) resolved_record_id = normalize_positive_id_int(locator["record_id"], field_name="record_id") resolved_workflow_node_id = int(locator["workflow_node_id"]) resolved_task_box = str(locator.get("task_box") or "todo") explicit_app_key = (app_key or "").strip() if explicit_app_key and explicit_app_key != resolved_app_key: raise_tool_error( QingflowApiError.config_error( f"task_id={task_id_text} resolved to app_key='{resolved_app_key}', which does not match app_key='{explicit_app_key}'" ) ) if record_id not in (None, ""): explicit_record_id = normalize_positive_id_text(record_id, field_name="record_id") if explicit_record_id != stringify_backend_id(resolved_record_id): raise_tool_error( QingflowApiError.config_error( f"task_id={task_id_text} resolved to record_id={resolved_record_id}, which does not match record_id={explicit_record_id}" ) ) if workflow_node_id not in (None, 0) and int(workflow_node_id) != resolved_workflow_node_id: raise_tool_error( QingflowApiError.config_error( f"task_id={task_id_text} resolved to workflow_node_id={resolved_workflow_node_id}, which does not match workflow_node_id={workflow_node_id}" ) ) else: resolved_record_id = normalize_positive_id_int(record_id, field_name="record_id") resolved_workflow_node_id = int(workflow_node_id) resolved_task_box = "todo" return { "task_id": task_id_text, "task_box": resolved_task_box, "app_key": resolved_app_key, "record_id": resolved_record_id, "record_id_text": stringify_backend_id(resolved_record_id), "workflow_node_id": resolved_workflow_node_id, } def _task_item_matches_query(self, item: dict[str, Any], query: str) -> bool: needle = str(query or "").strip().casefold() if not needle: return False for candidate in ( item.get("app_name"), item.get("workflow_node_name"), item.get("app_key"), item.get("record_id"), ): if candidate in (None, ""): continue if needle in str(candidate).casefold(): return True return False def task_associated_report_detail_get( self, *, profile: str, task_id: Any = None, app_key: str = "", record_id: Any = "", workflow_node_id: int = 0, report_id: int, page: int, page_size: int, ) -> dict[str, Any]: """执行任务相关逻辑。""" if task_id in (None, ""): normalize_positive_id_int(record_id, field_name="record_id") if report_id <= 0: raise_tool_error(QingflowApiError.config_error("report_id must be positive")) if page <= 0 or page_size <= 0: raise_tool_error(QingflowApiError.config_error("page and page_size must be positive")) def runner(session_profile, context): locator = self._resolve_task_locator_input( profile=profile, task_id=task_id, app_key=app_key, record_id=record_id, workflow_node_id=workflow_node_id, ) task_id_text = locator["task_id"] resolved_app_key = str(locator["app_key"]) resolved_record_id = int(locator["record_id"]) record_id_text = str(locator["record_id_text"] or "") resolved_workflow_node_id = int(locator["workflow_node_id"]) resolved_task_box = str(locator.get("task_box") or "todo") self._require_app_record_and_node(resolved_app_key, resolved_record_id, resolved_workflow_node_id) task_context = self._build_task_context( profile=profile, context=context, app_key=resolved_app_key, record_id=resolved_record_id, workflow_node_id=resolved_workflow_node_id, task_box=resolved_task_box, include_candidates=False, include_associated_reports=True, current_uid=session_profile.uid, ) report_item = self._find_associated_report(task_context, report_id) if report_item is None: raise_tool_error( QingflowApiError.config_error( f"report_id={report_id} is not visible for app_key='{resolved_app_key}' record_id={record_id_text} workflow_node_id={resolved_workflow_node_id}" ) ) association_query = self._build_association_query( report_item["raw"], task_context.get("record", {}).get("answers") or [], ) selection = { "app_key": resolved_app_key, "record_id": record_id_text, "workflow_node_id": resolved_workflow_node_id, "report_id": report_id, "target_app_key": report_item.get("target_app_key"), "target_app_name": report_item.get("target_app_name"), "chart_key": report_item.get("chart_key"), "chart_name": report_item.get("chart_name"), } if task_id_text is not None: selection["task_id"] = task_id_text context_payload = { "match_rules": report_item.get("match_rules") or [], "resolved_filters": association_query.get("keyQueValues") or [], } if report_item.get("graph_type") == "view": viewgraph_key = str(report_item.get("chart_key") or "") body = { "filter": {}, "viewgraphKey": viewgraph_key, "equipmentType": 0, "associationQuery": association_query, } result = self.backend.request( "POST", context, f"/view/{viewgraph_key}/apply/filter", json_body=body, ) items = _task_page_items(result) return { "profile": profile, "ws_id": session_profile.selected_ws_id, "ok": True, "request_route": self._request_route_payload(context), "warnings": [], "output_profile": "normal", "data": { "result_type": "view_list", "result": { "items": items, "pagination": { "page": page, "page_size": page_size, "returned_items": len(items), "page_amount": _task_page_amount(result), "reported_total": _task_page_total(result), }, }, "selection": selection, "context": context_payload, }, } chart_key = str(report_item.get("chart_key") or "") source_type = report_item.get("source_type") if source_type in {"qingbi", "bi_qingflow", "bi_dataset"}: qingbi_context = BackendRequestContext( base_url=_qingbi_base_url(context.base_url), token=context.token, ws_id=context.ws_id, qf_request_id=context.qf_request_id, qf_version=context.qf_version, qf_version_source=context.qf_version_source, ) chart_payload = { "asosChartId": report_id, "keyQueValues": association_query.get("keyQueValues") or [], } chart_params = { "qfUUID": uuid4().hex, "pageNum": page, "pageSize": page_size, } try: chart_result = self.backend.request( "POST", qingbi_context, f"/qingbi/charts/data/qflow/{chart_key}/detail", params=chart_params, json_body=chart_payload, ) except QingflowApiError as error: if not _should_retry_asos_data(error): raise try: chart_result = self.backend.request( "POST", qingbi_context, f"/qingbi/charts/data/qflow/{chart_key}/asos", params=chart_params, json_body=chart_payload, ) except QingflowApiError as fallback_error: if not _should_retry_asos_data(fallback_error): raise chart_result = self.backend.request( "POST", qingbi_context, f"/qingbi/charts/data/{chart_key}", params=chart_params, json_body=chart_payload, ) request_route = { "base_url": qingbi_context.base_url, "qf_version": qingbi_context.qf_version, "qf_version_source": qingbi_context.qf_version_source or "context", } elif self._qingflow_chart_uses_apply_filter(context, chart_key): chart_result = self.backend.request( "POST", context, f"/chart/{chart_key}/apply/filter", json_body={ "filter": { "pageNum": page, "pageSize": page_size, }, "asosChartId": report_id, "keyQueValues": association_query.get("keyQueValues") or [], }, ) items = _task_page_items(chart_result) return { "profile": profile, "ws_id": session_profile.selected_ws_id, "ok": True, "request_route": self._request_route_payload(context), "warnings": [], "output_profile": "normal", "data": { "result_type": "view_list", "result": { "items": items, "pagination": { "page": page, "page_size": page_size, "returned_items": len(items), "page_amount": _task_page_amount(chart_result), "reported_total": _task_page_total(chart_result), }, }, "selection": selection, "context": context_payload, }, } else: chart_result = self.backend.request( "POST", context, f"/chart/{chart_key}/chartData", json_body={ "asosChartId": report_id, "keyQueValues": association_query.get("keyQueValues") or [], }, ) request_route = self._request_route_payload(context) return { "profile": profile, "ws_id": session_profile.selected_ws_id, "ok": True, "request_route": request_route, "warnings": [], "output_profile": "normal", "data": { "result_type": "chart_data", "result": self._normalize_chart_result(chart_result), "selection": selection, "context": context_payload, }, } return self._run(profile, runner, tool_name='任务关联报表详情') def task_workflow_log_get( self, *, profile: str, task_id: Any = None, app_key: str = "", record_id: Any = "", workflow_node_id: int = 0, ) -> dict[str, Any]: """执行任务相关逻辑。""" if task_id in (None, ""): normalize_positive_id_int(record_id, field_name="record_id") def runner(session_profile, context): locator = self._resolve_task_locator_input( profile=profile, task_id=task_id, app_key=app_key, record_id=record_id, workflow_node_id=workflow_node_id, ) task_id_text = locator["task_id"] resolved_app_key = str(locator["app_key"]) resolved_record_id = int(locator["record_id"]) record_id_text = str(locator["record_id_text"] or "") resolved_workflow_node_id = int(locator["workflow_node_id"]) resolved_task_box = str(locator.get("task_box") or "todo") self._require_app_record_and_node(resolved_app_key, resolved_record_id, resolved_workflow_node_id) task_context = self._build_task_context( profile=profile, context=context, app_key=resolved_app_key, record_id=resolved_record_id, workflow_node_id=resolved_workflow_node_id, task_box=resolved_task_box, include_candidates=False, include_associated_reports=False, current_uid=session_profile.uid, ) visibility = task_context.get("visibility") or {} node = task_context.get("node") if isinstance(task_context.get("node"), dict) else {} raw_node = node.get("raw") if isinstance(node.get("raw"), dict) else {} audit_visibility_explicit = "auditRecordVisible" in raw_node if audit_visibility_explicit and not visibility.get("audit_record_visible"): raise_tool_error( QingflowApiError.config_error( f"workflow logs are not visible for app_key='{resolved_app_key}' record_id={record_id_text} workflow_node_id={resolved_workflow_node_id}" ) ) page = self.backend.request( "POST", context, "/application/workflow/node/record", json_body={ "key": resolved_app_key, "rowRecordId": resolved_record_id, "nodeId": resolved_workflow_node_id, "role": _task_box_record_role(resolved_task_box), "pageNum": 1, "pageSize": 200, }, ) items = self._normalize_workflow_logs(page) result = { "profile": profile, "ws_id": session_profile.selected_ws_id, "ok": True, "request_route": self._request_route_payload(context), "warnings": task_context.get("warnings") or [], "output_profile": "normal", "data": { "selection": { "app_key": resolved_app_key, "record_id": record_id_text, "workflow_node_id": resolved_workflow_node_id, }, "visibility": { "audit_record_visible": ( visibility.get("audit_record_visible") if audit_visibility_explicit else None ), "qrobot_record_visible": visibility.get("qrobot_record_visible"), }, "items": items, }, } if task_id_text is not None: selection = result["data"].get("selection") if isinstance(selection, dict): selection["task_id"] = task_id_text return result return self._run(profile, runner, tool_name='任务流程日志') def _build_task_context( self, profile: str, context: BackendRequestContext, *, app_key: str, record_id: int, workflow_node_id: int, include_candidates: bool, include_associated_reports: bool, current_uid: int | None = None, task_box: str = "todo", ) -> dict[str, Any]: """执行内部辅助逻辑。""" context_warnings: list[JSONObject] = [] detail: dict[str, Any] | None = None list_type = _task_box_record_list_type(task_box) role = _task_box_record_role(task_box) try: audit_infos = self.backend.request( "GET", context, f"/app/{app_key}/apply/{record_id}/auditInfo", params={"type": list_type}, ) node_info = self._select_task_node(audit_infos, workflow_node_id, app_key=app_key, record_id=record_id) except QingflowApiError as exc: if not _is_optional_task_audit_info_error(exc): raise detail = self.backend.request( "GET", context, f"/app/{app_key}/apply/{record_id}", params={"role": role, "listType": list_type, "auditNodeId": workflow_node_id}, ) node_info = self._fallback_task_node_from_detail( detail, workflow_node_id=workflow_node_id, source_error=exc, ) context_warnings.append( { "code": "TASK_AUDIT_INFO_UNAVAILABLE", "message": "Task context used the current task detail because auditInfo was unavailable in this permission context.", "backend_code": exc.backend_code, "request_id": exc.request_id, "http_status": exc.http_status, } ) if detail is None: detail = self.backend.request( "GET", context, f"/app/{app_key}/apply/{record_id}", params={"role": role, "listType": list_type, "auditNodeId": workflow_node_id}, ) app_name = self._task_app_name(context=context, app_key=app_key, detail=detail, node_info=node_info) associated_report_visible = self._resolve_associated_report_visible(node_info, detail) associated_reports = {"visible": associated_report_visible, "loaded": False, "count": 0, "items": [], "warnings": []} if include_associated_reports and associated_report_visible: try: asos_chart_list = self.backend.request( "GET", context, f"/app/{app_key}/asosChart", params={"role": role, "auditNodeId": workflow_node_id, "beingDraft": False}, ) associated_items = [ self._normalize_associated_report(item) for item in (asos_chart_list.get("asosCharts") or []) if isinstance(item, dict) ] associated_reports = { "visible": True, "loaded": True, "count": len(associated_items), "items": associated_items, "warnings": [], } except QingflowApiError as exc: if not _is_task_optional_read_error(exc): raise associated_reports = { "visible": True, "loaded": False, "count": 0, "items": [], "warnings": [ { "code": "TASK_ASSOCIATED_REPORTS_UNAVAILABLE", "message": "Associated reports are not readable in this permission context; task detail remains available.", "backend_code": exc.backend_code, "request_id": exc.request_id, "http_status": exc.http_status, } ], } rollback_items: list[dict[str, Any]] = [] rollback_warnings: list[JSONObject] = [] transfer_items: list[dict[str, Any]] = [] transfer_warnings: list[JSONObject] = [] transfer_pagination: JSONObject = { "loaded": False, "page_size": 100, "fetched_pages": 0, "reported_total": None, "page_amount": None, "truncated": False, } if include_candidates: try: rollback_result = self.backend.request( "GET", context, f"/app/{app_key}/apply/{record_id}/revertNode", params={"auditNodeId": workflow_node_id}, ) rollback_items = self._rollback_candidate_items(rollback_result) except QingflowApiError as exc: if not _is_task_optional_read_error(exc): raise rollback_warnings.append( { "code": "TASK_ROLLBACK_CANDIDATES_UNAVAILABLE", "message": "Rollback candidates are not readable in this permission context; task detail remains available.", "backend_code": exc.backend_code, "request_id": exc.request_id, "http_status": exc.http_status, } ) transfer_items, transfer_warnings, transfer_pagination = self._transfer_candidate_items( context, app_key=app_key, record_id=record_id, workflow_node_id=workflow_node_id, current_uid=current_uid, ) update_schema_state = self._build_task_update_schema( profile=profile, context=context, app_key=app_key, record_id=record_id, workflow_node_id=workflow_node_id, node_info=node_info, current_answers=detail.get("answers") or [], ) update_schema = update_schema_state["public_schema"] save_only_available, capability_warnings, save_only_source = self._resolve_task_save_only_availability( context, app_key=app_key, workflow_node_id=workflow_node_id, node_info=node_info, ) capabilities = self._build_capabilities( node_info, allow_save_only=save_only_available, warnings=capability_warnings, save_only_source=save_only_source, ) visibility = self._build_visibility(node_info, detail) record_id_text = stringify_backend_id(record_id) return { "task": { "app_key": app_key, "app_name": app_name, "record_id": record_id_text, "workflow_node_id": workflow_node_id, "workflow_node_name": node_info.get("auditNodeName") or node_info.get("nodeName"), "actionable": True, }, "node": { "workflow_node_id": workflow_node_id, "workflow_node_name": node_info.get("auditNodeName") or node_info.get("nodeName"), "raw": dict(node_info), }, "record": { "apply_id": stringify_backend_id(detail.get("applyId") or record_id), "apply_status": detail.get("applyStatus"), "apply_num": detail.get("applyNum"), "custom_apply_num": detail.get("customApplyNum"), "apply_user": detail.get("applyUser"), "apply_time": detail.get("applyTime"), "last_update_time": detail.get("lastUpdateTime"), "answers": detail.get("answers") or [], }, "capabilities": capabilities, "field_permissions": { "que_auth_setting": node_info.get("queAuthSetting") or [], "editable_question_ids": update_schema_state["editable_question_ids"], "editable_question_ids_source": update_schema_state["editable_question_ids_source"], }, "visibility": visibility, "associated_reports": associated_reports, "candidates": { "rollback_nodes": rollback_items, "transfer_members": transfer_items, "loaded": include_candidates, "transfer_pagination": transfer_pagination, "rollback_warnings": rollback_warnings, "transfer_warnings": transfer_warnings, "warnings": [*rollback_warnings, *transfer_warnings], }, "workflow_log_summary": { "visible": visibility["audit_record_visible"], "available": visibility["audit_record_visible"], "history_count": None, "qrobot_log_visible": visibility["qrobot_record_visible"], }, "update_schema": update_schema, "warnings": context_warnings, } def _compact_task_get_context(self, data: dict[str, Any]) -> dict[str, Any]: task = data.get("task") if isinstance(data.get("task"), dict) else {} record = data.get("record") if isinstance(data.get("record"), dict) else {} capabilities = data.get("capabilities") if isinstance(data.get("capabilities"), dict) else {} update_schema = data.get("update_schema") if isinstance(data.get("update_schema"), dict) else {} associated_reports = data.get("associated_reports") if isinstance(data.get("associated_reports"), dict) else {} candidates = data.get("candidates") if isinstance(data.get("candidates"), dict) else {} workflow_log = data.get("workflow_log_summary") if isinstance(data.get("workflow_log_summary"), dict) else {} available_actions = [ str(item) for item in (capabilities.get("available_actions") or []) if str(item).strip() ] writable_fields = update_schema.get("writable_fields") if isinstance(update_schema.get("writable_fields"), list) else [] rollback_items = [ self._compact_rollback_candidate(item) for item in (candidates.get("rollback_nodes") or []) if isinstance(item, dict) ] transfer_items = [ self._compact_transfer_member(item) for item in (candidates.get("transfer_members") or []) if isinstance(item, dict) ] associated_items = [ self._compact_associated_report(item) for item in (associated_reports.get("items") or []) if isinstance(item, dict) ] transfer_pagination = candidates.get("transfer_pagination") if isinstance(candidates.get("transfer_pagination"), dict) else {} compact: dict[str, Any] = { "task": { "app_key": task.get("app_key"), "app_name": task.get("app_name"), "record_id": stringify_backend_id(task.get("record_id")), "workflow_node_id": task.get("workflow_node_id"), "workflow_node_name": task.get("workflow_node_name"), "initiator": self._compact_initiator(record.get("apply_user")), "actionable": task.get("actionable"), }, "record_summary": { "apply_status": record.get("apply_status"), "apply_num": record.get("apply_num"), "custom_apply_num": record.get("custom_apply_num"), "apply_time": record.get("apply_time"), "last_update_time": record.get("last_update_time"), "core_fields": self._task_record_core_fields(record.get("answers") or []), "all_fields": self._task_record_all_fields(record.get("answers") or []), }, "available_actions": available_actions, "editable_fields": [ self._compact_task_editable_field(item, update_schema) for item in writable_fields if isinstance(item, dict) ], "extras": { "workflow_log": { "available": bool(workflow_log.get("available")), "qrobot_log_visible": bool(workflow_log.get("qrobot_log_visible")), "history_count": workflow_log.get("history_count"), }, "associated_reports": { "available": bool(associated_reports.get("visible")), "loaded": bool(associated_reports.get("loaded")), "count": len(associated_items), "items": associated_items, "warnings": associated_reports.get("warnings") or [], }, "rollback_candidates": { "available": "rollback" in available_actions, "loaded": bool(candidates.get("loaded")), "count": len(rollback_items), "items": rollback_items, "warnings": candidates.get("rollback_warnings") or [], }, "transfer_candidates": { "available": "transfer" in available_actions, "loaded": bool(transfer_pagination.get("loaded")), "count": len(transfer_items), "items": transfer_items, "pagination": transfer_pagination, "warnings": candidates.get("transfer_warnings") or [], }, }, } action_metadata = self._compact_task_action_metadata(capabilities) if action_metadata: compact["action_metadata"] = action_metadata editable_metadata = self._compact_task_editable_metadata(update_schema) if editable_metadata: compact["editable_metadata"] = editable_metadata return compact def _compact_task_action_metadata(self, capabilities: dict[str, Any]) -> dict[str, Any]: constraints = capabilities.get("action_constraints") if isinstance(capabilities.get("action_constraints"), dict) else {} metadata: dict[str, Any] = {} feedback_required_for = constraints.get("feedback_required_for") if isinstance(constraints.get("feedback_required_for"), list) else [] if feedback_required_for: metadata["feedback_required_for"] = feedback_required_for visible_but_unimplemented = capabilities.get("visible_but_unimplemented_actions") if visible_but_unimplemented: metadata["visible_but_unimplemented_actions"] = visible_but_unimplemented if capabilities.get("save_only_source"): metadata["save_only_source"] = capabilities.get("save_only_source") if capabilities.get("warnings"): metadata["warnings"] = capabilities.get("warnings") return metadata def _compact_task_editable_metadata(self, update_schema: dict[str, Any]) -> dict[str, Any]: metadata: dict[str, Any] = {} blockers = update_schema.get("blockers") if isinstance(update_schema.get("blockers"), list) else [] warnings = update_schema.get("warnings") if isinstance(update_schema.get("warnings"), list) else [] if blockers: metadata["blockers"] = blockers if warnings: metadata["warnings"] = warnings if update_schema.get("editable_question_ids_source"): metadata["editable_question_ids_source"] = update_schema.get("editable_question_ids_source") return metadata def _compact_initiator(self, payload: Any) -> dict[str, Any] | None: if not isinstance(payload, dict): return None compact = { "uid": payload.get("uid"), "displayName": payload.get("displayName") or payload.get("name") or payload.get("nickName"), "email": payload.get("email"), "mobile": payload.get("mobile"), "headImg": payload.get("headImg"), } return {key: value for key, value in compact.items() if value not in (None, "", [])} or None def _task_app_name( self, *, context: BackendRequestContext, app_key: str, detail: dict[str, Any], node_info: dict[str, Any], ) -> Any: for source in (detail, node_info): for key in ("formTitle", "appName", "worksheetName", "appTitle"): value = source.get(key) if value not in (None, ""): if app_key: self._app_name_cache[app_key] = str(value) return value normalized_app_key = str(app_key or "").strip() if not normalized_app_key: return None if normalized_app_key in self._app_name_cache: return self._app_name_cache[normalized_app_key] resolved = self._resolve_task_app_name_from_base_info(context=context, app_key=normalized_app_key) if resolved is None: resolved = self._resolve_task_app_name_from_visible_apps(context=context, app_key=normalized_app_key) self._app_name_cache[normalized_app_key] = resolved return resolved def _resolve_task_app_name_from_base_info( self, *, context: BackendRequestContext, app_key: str, ) -> str | None: try: base_info = self.backend.request("GET", context, f"/app/{app_key}/baseInfo") except QingflowApiError as exc: if not _is_task_optional_read_error(exc): raise return None if not isinstance(base_info, dict): return None for key in ("formTitle", "title", "appName", "name"): value = str(base_info.get(key) or "").strip() if value: return value return None def _resolve_task_app_name_from_visible_apps( self, *, context: BackendRequestContext, app_key: str, ) -> str | None: try: visible_apps = self.backend.request("GET", context, "/tag/apps") except QingflowApiError as exc: if not _is_task_optional_read_error(exc): raise return None return self._find_task_app_name_in_visible_apps(visible_apps, app_key=app_key) def _find_task_app_name_in_visible_apps(self, payload: Any, *, app_key: str) -> str | None: if isinstance(payload, list): for item in payload: resolved = self._find_task_app_name_in_visible_apps(item, app_key=app_key) if resolved: return resolved return None if not isinstance(payload, dict): return None candidate_app_key = str(payload.get("appKey") or payload.get("app_key") or "").strip() if candidate_app_key == app_key: for key in ("formTitle", "title", "appName", "name"): value = str(payload.get(key) or "").strip() if value: return value for value in payload.values(): if isinstance(value, (list, dict)): resolved = self._find_task_app_name_in_visible_apps(value, app_key=app_key) if resolved: return resolved return None def _task_record_core_fields(self, answers: Any, *, limit: int = 12) -> dict[str, Any]: return self._task_record_field_map(answers, limit=limit, truncate_text=160) def _task_record_all_fields(self, answers: Any) -> dict[str, Any]: return self._task_record_field_map(answers, limit=None, truncate_text=None) def _task_record_field_map( self, answers: Any, *, limit: int | None, truncate_text: int | None, ) -> dict[str, Any]: if not isinstance(answers, list): return {} field_map: dict[str, Any] = {} for answer in answers: if not isinstance(answer, dict): continue title = answer.get("queTitle") or answer.get("title") or answer.get("fieldName") if not title: que_id = answer.get("queId") title = f"field_{que_id}" if que_id not in (None, "") else None if not title: continue table_values = answer.get("tableValues") if isinstance(answer.get("tableValues"), list) else [] if table_values: value: Any = f"子表格 {len(table_values)} 行" else: values = self._extract_answer_values(answer) if not values: continue value = values[0] if len(values) == 1 else values if value in (None, "", []): continue field_map[str(title)] = self._compact_task_value(value, truncate_text=truncate_text) if limit is not None and len(field_map) >= limit: break return field_map def _compact_task_value(self, value: Any, *, truncate_text: int | None = 160) -> Any: if isinstance(value, list): items = [self._compact_task_value(item, truncate_text=truncate_text) for item in value] if truncate_text is not None: return items[:8] return items text = re.sub(r"<[^>]+>", " ", str(value)) text = re.sub(r"\s+", " ", text).strip() if truncate_text is None or len(text) <= truncate_text: return text if truncate_text <= 3: return text[:truncate_text] return text[: truncate_text - 3].rstrip() + "..." def _compact_task_editable_field(self, field: dict[str, Any], update_schema: dict[str, Any]) -> dict[str, Any]: payload_template = update_schema.get("payload_template") if isinstance(update_schema.get("payload_template"), dict) else {} title = field.get("title") compact: dict[str, Any] = {} for key in ("field_id", "title", "kind", "required", "candidate_hint"): if key in field: compact[key] = field.get(key) if title in payload_template: compact["template"] = payload_template.get(title) return compact def _compact_associated_report(self, item: dict[str, Any]) -> dict[str, Any]: return { key: value for key, value in { "report_id": item.get("report_id"), "chart_key": item.get("chart_key"), "chart_name": item.get("chart_name"), "graph_type": item.get("graph_type"), "source_type": item.get("source_type"), "target_app_key": item.get("target_app_key"), "target_app_name": item.get("target_app_name"), }.items() if value not in (None, "", []) } def _compact_rollback_candidate(self, item: dict[str, Any]) -> dict[str, Any]: return { key: value for key, value in { "workflow_node_id": item.get("auditNodeId") or item.get("nodeId"), "workflow_node_name": item.get("auditNodeName") or item.get("nodeName"), }.items() if value not in (None, "", []) } def _compact_transfer_member(self, item: dict[str, Any]) -> dict[str, Any]: uid = item.get("uid") if uid is None: uid = item.get("userId") or item.get("memberId") or item.get("id") return { key: value for key, value in { "uid": uid, "name": item.get("name") or item.get("userName") or item.get("memberName") or item.get("realName"), "email": item.get("email") or item.get("mail"), "department_id": item.get("departmentId") or item.get("deptId"), "department_name": item.get("departmentName") or item.get("deptName"), }.items() if value not in (None, "", []) } def _normalize_task_item(self, raw: dict[str, Any]) -> dict[str, Any]: """执行内部辅助逻辑。""" app_key = raw.get("appKey") or raw.get("app_key") record_id = raw.get("rowRecordId") or raw.get("recordId") or raw.get("applyId") workflow_node_id = raw.get("nodeId") or raw.get("auditNodeId") return { "task_id": stringify_backend_id(raw.get("id") or raw.get("taskId") or record_id), "app_key": app_key, "app_name": raw.get("formTitle") or raw.get("worksheetName") or raw.get("appName"), "record_id": stringify_backend_id(record_id), "workflow_node_id": workflow_node_id, "workflow_node_name": raw.get("nodeName") or raw.get("auditNodeName"), "apply_time": raw.get("applyTime") or raw.get("receiveTime"), "summary_fields": self._normalize_task_summary_fields(raw.get("dataSnapshot")), } def _public_task_item(self, item: dict[str, Any]) -> dict[str, Any]: return { "task_id": item.get("task_id"), "app_name": item.get("app_name"), "workflow_node_name": item.get("workflow_node_name"), "apply_time": item.get("apply_time"), "summary_fields": item.get("summary_fields") if isinstance(item.get("summary_fields"), list) else [], } def _normalize_task_summary_fields(self, raw: Any) -> list[dict[str, Any]]: """执行内部辅助逻辑。""" if not isinstance(raw, list): return [] summary_fields: list[dict[str, Any]] = [] for item in raw: if not isinstance(item, dict): continue summary_field: dict[str, Any] = { "field_id": item.get("fieldId"), "title": item.get("fieldTitle"), "type": item.get("fieldType"), "answer": item.get("fieldAnswer"), "desensitized": self._coerce_bool(item.get("beingDesensitized")), } associated_field_type = item.get("associatedQueType") if associated_field_type is not None: summary_field["associated_field_type"] = associated_field_type summary_fields.append(summary_field) return summary_fields def _select_task_node(self, infos: Any, workflow_node_id: int, *, app_key: str, record_id: int) -> dict[str, Any]: """执行内部辅助逻辑。""" if not isinstance(infos, list) or not infos: raise_tool_error( QingflowApiError.config_error( f"record_id={record_id} is not currently actionable for the logged-in user in app_key='{app_key}'" ) ) for item in infos: if not isinstance(item, dict): continue candidate = item.get("auditNodeId") if not isinstance(candidate, int): candidate = item.get("nodeId") if candidate == workflow_node_id: return item raise_tool_error( QingflowApiError.config_error( f"workflow_node_id={workflow_node_id} is not an actionable todo node for app_key='{app_key}' record_id={record_id}" ) ) def _fallback_task_node_from_detail( self, detail: dict[str, Any], *, workflow_node_id: int, source_error: QingflowApiError, ) -> dict[str, Any]: """Build a conservative task node snapshot from the readable task detail.""" node_info: dict[str, Any] = { "auditNodeId": workflow_node_id, "nodeId": workflow_node_id, "auditNodeName": ( detail.get("auditNodeName") or detail.get("nodeName") or detail.get("workflowNodeName") or detail.get("currentNodeName") ), "_auditInfoUnavailable": True, "_auditInfoError": { "http_status": source_error.http_status, "backend_code": source_error.backend_code, "category": source_error.category, }, } for key in ( "canTransfer", "canRevert", "canUrge", "rejectBtnStatus", "canRevoke", "beingEndWorkflow", "beingCanApplyAgain", "beingSubmitCheck", "beingSubmitPreview", "feedbackRequiredOperationType", "queAuthSetting", "auditRecordVisible", "qrobotRecordBeingVisible", "beingWorkflowNodeFutureListVisible", "commentStatus", "asosChartVisible", ): if key in detail: node_info[key] = detail[key] if "viewAsosChartVisible" in detail and "asosChartVisible" not in node_info: node_info["asosChartVisible"] = detail.get("viewAsosChartVisible") return node_info def _build_capabilities( self, node_info: dict[str, Any], *, allow_save_only: bool, warnings: list[JSONObject] | None = None, save_only_source: str = "workflow_editable_que_ids", ) -> dict[str, Any]: """执行内部辅助逻辑。""" available_actions = ["approve"] if self._coerce_bool(node_info.get("rejectBtnStatus")): available_actions.append("reject") if self._coerce_bool(node_info.get("canRevert")): available_actions.append("rollback") if self._coerce_bool(node_info.get("canTransfer")): available_actions.append("transfer") if self._coerce_bool(node_info.get("canUrge")): available_actions.append("urge") if allow_save_only: available_actions.append("save_only") visible_but_unimplemented_actions: list[str] = [] if self._coerce_bool(node_info.get("canRevoke")): visible_but_unimplemented_actions.append("revoke") if self._coerce_bool(node_info.get("beingEndWorkflow")): visible_but_unimplemented_actions.append("end_workflow") if self._coerce_bool(node_info.get("beingCanApplyAgain")): visible_but_unimplemented_actions.append("apply_again") feedback_required_for = [] raw_feedback_required = node_info.get("feedbackRequiredOperationType") if isinstance(raw_feedback_required, list): feedback_required_for = [str(item).strip().lower() for item in raw_feedback_required if str(item).strip()] return { "available_actions": available_actions, "visible_but_unimplemented_actions": visible_but_unimplemented_actions, "save_only_source": save_only_source, "warnings": list(warnings or []), "action_constraints": { "feedback_required_for": feedback_required_for, "submit_check_enabled": self._coerce_bool(node_info.get("beingSubmitCheck")), "submit_preview_enabled": self._coerce_bool(node_info.get("beingSubmitPreview")), "can_end_workflow": self._coerce_bool(node_info.get("beingEndWorkflow")), "can_apply_again": self._coerce_bool(node_info.get("beingCanApplyAgain")), }, } def _build_task_update_schema( self, profile: str, context: BackendRequestContext, *, app_key: str, record_id: int, workflow_node_id: int, node_info: dict[str, Any], current_answers: Any, ) -> dict[str, Any]: """执行内部辅助逻辑。""" record_id_text = stringify_backend_id(record_id) schema_warnings: list[JSONObject] = [] try: app_schema = _normalize_form_schema( self.backend.request( "GET", context, f"/app/{app_key}/form", params={"type": 3, "auditNodeId": workflow_node_id}, ) ) except QingflowApiError as error: if not _is_task_optional_read_error(error): raise schema_warnings.append( { "code": "TASK_NODE_UPDATE_SCHEMA_UNAVAILABLE", "message": "task update schema fell back to the applicant form because the processing-node form was unavailable in this permission context.", "backend_code": error.backend_code, "http_status": error.http_status, "request_id": error.request_id, } ) try: app_schema = self._record_tools._get_form_schema(profile, context, app_key, force_refresh=False) except QingflowApiError as fallback_error: if not _is_task_optional_read_error(fallback_error): raise return self._build_task_update_schema_from_runtime_answers( profile=profile, context=context, app_key=app_key, record_id=record_id, workflow_node_id=workflow_node_id, node_info=node_info, current_answers=current_answers, source_error=fallback_error, ) question_relations = _collect_question_relations(app_schema) linked_field_ids = _collect_linked_required_field_ids(question_relations) base_index = _build_applicant_top_level_field_index(app_schema) linked_field_ids.update(_collect_option_linked_field_ids(base_index)) linked_hidden_index = _build_applicant_hidden_linked_top_level_field_index( app_schema, linked_field_ids=linked_field_ids, ) index = _merge_field_indexes(base_index, linked_hidden_index) editable_question_ids, editable_warnings, source = self._resolve_task_editable_question_ids( context, app_key=app_key, workflow_node_id=workflow_node_id, node_info=node_info, ) schema_warnings.extend(editable_warnings) index, augmentation_warnings = self._augment_task_editable_field_index( index=index, current_answers=current_answers, editable_question_ids=editable_question_ids, ) schema_warnings.extend(augmentation_warnings) effective_editable_ids = set(editable_question_ids) for field in index.by_id.values(): if field.que_type in SUBTABLE_QUE_TYPES and (_subtable_descendant_ids(field) & set(editable_question_ids)): effective_editable_ids.add(field.que_id) writable_fields: list[JSONObject] = [] linkage_payloads_by_field_id = _build_static_schema_linkage_payloads( index=index, question_relations=question_relations, ) for field in index.by_id.values(): if field.que_type in LAYOUT_ONLY_QUE_TYPES or field.que_id not in effective_editable_ids: continue editable_field = _clone_form_field(field, readonly=False) write_hints = self._record_tools._schema_write_hints(editable_field) if not bool(write_hints.get("writable")): continue writable_field = self._record_tools._ready_schema_field_payload( profile, context, editable_field, ws_id=context.ws_id, required_override=False, linkage_payloads_by_field_id=linkage_payloads_by_field_id, ) writable_field.setdefault("field_id", editable_field.que_id) writable_fields.append(writable_field) blockers: list[str] = [] if not writable_fields: blockers.append("NO_TASK_EDITABLE_FIELDS") schema_warnings.append( { "code": "NO_TASK_EDITABLE_FIELDS", "message": "the current task node does not expose any writable fields for task-scoped edits.", } ) public_schema: JSONObject = { "schema_scope": "task_update_ready", "writable_fields": writable_fields, "payload_template": { item["title"]: self._record_tools._ready_schema_template_value(item) for item in writable_fields if isinstance(item, dict) and item.get("title") }, "blockers": blockers, "warnings": schema_warnings, "selection": { "app_key": app_key, "record_id": record_id_text, "workflow_node_id": workflow_node_id, }, "editable_question_ids_source": source, } return { "public_schema": public_schema, "index": index, "editable_question_ids": sorted(editable_question_ids), "effective_editable_question_ids": sorted(effective_editable_ids), "editable_question_ids_source": source, } def _build_task_update_schema_from_runtime_answers( self, *, profile: str, context: BackendRequestContext, app_key: str, record_id: int, workflow_node_id: int, node_info: dict[str, Any], current_answers: Any, source_error: QingflowApiError, ) -> dict[str, Any]: """Build node-scoped edit schema from the task detail when app schema is unavailable.""" record_id_text = stringify_backend_id(record_id) editable_question_ids, schema_warnings, source = self._resolve_task_editable_question_ids( context, app_key=app_key, workflow_node_id=workflow_node_id, node_info=node_info, ) schema_warnings.insert( 0, { "code": "TASK_UPDATE_SCHEMA_APP_SCHEMA_UNAVAILABLE", "message": "task update schema used current task answers because the app applicant schema was unavailable in this permission context.", "transport_error": { "http_status": source_error.http_status, "backend_code": source_error.backend_code, "category": source_error.category, }, }, ) index = _build_answer_backed_field_index( current_answers, field_id_filter={str(que_id) for que_id in editable_question_ids if que_id > 0}, ) effective_editable_ids = set(editable_question_ids) writable_fields: list[JSONObject] = [] for field in index.by_id.values(): if field.que_type in LAYOUT_ONLY_QUE_TYPES or field.que_id not in effective_editable_ids: continue editable_field = _clone_form_field(field, readonly=False) write_hints = self._record_tools._schema_write_hints(editable_field) if not bool(write_hints.get("writable")): continue writable_field = self._record_tools._ready_schema_field_payload( profile, context, editable_field, ws_id=context.ws_id, required_override=False, linkage_payloads_by_field_id={}, ) writable_field.setdefault("field_id", editable_field.que_id) writable_fields.append(writable_field) blockers: list[str] = [] if not writable_fields: blockers.append("NO_TASK_EDITABLE_FIELDS") schema_warnings.append( { "code": "NO_TASK_EDITABLE_FIELDS", "message": "the current task node does not expose any writable fields in the current task detail.", } ) public_schema: JSONObject = { "schema_scope": "task_update_ready", "writable_fields": writable_fields, "payload_template": { item["title"]: self._record_tools._ready_schema_template_value(item) for item in writable_fields if isinstance(item, dict) and item.get("title") }, "blockers": blockers, "warnings": schema_warnings, "selection": { "app_key": app_key, "record_id": record_id_text, "workflow_node_id": workflow_node_id, }, "editable_question_ids_source": f"{source}_runtime_answers", } return { "public_schema": public_schema, "index": index, "editable_question_ids": sorted(editable_question_ids), "effective_editable_question_ids": sorted(effective_editable_ids), "editable_question_ids_source": f"{source}_runtime_answers", } def _augment_task_editable_field_index( self, *, index: FieldIndex, current_answers: Any, editable_question_ids: set[int], ) -> tuple[FieldIndex, list[JSONObject]]: """执行内部辅助逻辑。""" if not editable_question_ids: return index, [] missing_field_ids = { str(que_id) for que_id in editable_question_ids if que_id > 0 and str(que_id) not in index.by_id } if not missing_field_ids: return index, [] answer_backed_index = _build_answer_backed_field_index( current_answers, field_id_filter=missing_field_ids, ) if not answer_backed_index.by_id: return index, [] augmented_index = _merge_field_indexes(index, answer_backed_index) return augmented_index, [ { "code": "TASK_RUNTIME_EDITABLE_FIELDS_AUGMENTED", "message": "task update schema added backend-editable fields from current task answers because applicant schema did not expose them.", "fields": [ { "field_id": field.que_id, "title": field.que_title, "que_type": field.que_type, } for field in answer_backed_index.by_id.values() ], } ] def _resolve_task_editable_question_ids( self, context: BackendRequestContext, *, app_key: str, workflow_node_id: int, node_info: dict[str, Any], ) -> tuple[set[int], list[JSONObject], str]: """执行内部辅助逻辑。""" warnings: list[JSONObject] = [] try: payload = self.backend.request( "GET", context, f"/app/{app_key}/auditNode/{workflow_node_id}/editableQueIds", ) question_ids = self._extract_question_ids(payload) if question_ids: return question_ids, warnings, "workflow_editable_que_ids" except QingflowApiError as error: if not _is_task_optional_read_error(error): raise warnings.append( { "code": "TASK_EDITABLE_IDS_FALLBACK", "message": "editable question ids endpoint is unavailable in the current route; task update schema fell back to queAuthSetting and may be conservative.", "backend_code": error.backend_code, "http_status": error.http_status, "request_id": error.request_id, } ) fallback_ids = self._editable_ids_from_que_auth_setting(node_info.get("queAuthSetting")) return fallback_ids, warnings, "que_auth_setting" def _resolve_task_save_only_availability( self, context: BackendRequestContext, *, app_key: str, workflow_node_id: int, node_info: dict[str, Any], ) -> tuple[bool, list[JSONObject], str]: """执行内部辅助逻辑。""" try: payload = self.backend.request( "GET", context, f"/app/{app_key}/auditNode/{workflow_node_id}/editableQueIds", ) except QingflowApiError as error: if not _is_task_optional_read_error(error): raise fallback_ids = self._editable_ids_from_que_auth_setting(node_info.get("queAuthSetting")) fallback_source = "que_auth_setting" if fallback_ids else "backend_editable_que_ids_unavailable" warning: JSONObject = { "code": "TASK_SAVE_ONLY_EDITABLE_IDS_FALLBACK" if fallback_ids else "TASK_SAVE_ONLY_SIGNAL_UNAVAILABLE", "message": ( "save_only availability used current task queAuthSetting because backend editableQueIds was unavailable in this permission context." if fallback_ids else "save_only is hidden because backend editableQueIds is unavailable and current task detail does not expose editable fields." ), } if error.backend_code is not None: warning["backend_code"] = error.backend_code if error.http_status is not None: warning["http_status"] = error.http_status if error.request_id is not None: warning["request_id"] = error.request_id if fallback_ids: warning["field_ids"] = sorted(fallback_ids) return True, [warning], fallback_source return False, [warning], fallback_source return bool(self._extract_question_ids(payload)), [], "workflow_editable_que_ids" def _extract_question_ids(self, payload: Any) -> set[int]: """执行内部辅助逻辑。""" candidates: list[Any] = [] if isinstance(payload, list): candidates = payload elif isinstance(payload, dict): for key in ("editableQueIds", "editableQuestionIds", "queIds", "questionIds", "ids", "list", "result"): value = payload.get(key) if isinstance(value, list): candidates = value break question_ids: set[int] = set() for item in candidates: if isinstance(item, int) and item > 0: question_ids.add(item) continue if isinstance(item, dict): for key in ("queId", "questionId", "id"): value = _coerce_count(item.get(key)) if value is not None and value > 0: question_ids.add(value) break return question_ids def _editable_ids_from_que_auth_setting(self, payload: Any) -> set[int]: """执行内部辅助逻辑。""" if not isinstance(payload, list): return set() editable_ids: set[int] = set() for item in payload: if not isinstance(item, dict): continue que_id = _coerce_count(item.get("queId") or item.get("questionId")) if que_id is None or que_id <= 0: continue explicit_editable_keys = ("editable", "writable", "canEdit", "editStatus", "beingEditable") explicit_readonly_keys = ("readonly", "beingReadonly") if any(key in item for key in explicit_editable_keys): if any(bool(item.get(key)) for key in explicit_editable_keys): editable_ids.add(que_id) continue if any(bool(item.get(key)) for key in explicit_readonly_keys): continue if item.get("readable") is False: continue if item.get("readable") is True or item.get("visible") is True: editable_ids.add(que_id) return editable_ids def _prepare_task_field_update( self, *, profile: str, context: BackendRequestContext, app_key: str, record_id: int, workflow_node_id: int, task_context: dict[str, Any], fields: dict[str, Any], ) -> dict[str, Any]: """执行内部辅助逻辑。""" record = task_context.get("record") if isinstance(task_context.get("record"), dict) else {} current_answers = record.get("answers") if isinstance(record.get("answers"), list) else [] node = task_context.get("node") if isinstance(task_context.get("node"), dict) else {} node_info = node.get("raw") if isinstance(node.get("raw"), dict) else {} schema_state = self._build_task_update_schema( profile=profile, context=context, app_key=app_key, record_id=record_id, workflow_node_id=workflow_node_id, node_info=node_info, current_answers=current_answers, ) update_schema = schema_state["public_schema"] if update_schema.get("blockers"): raise_tool_error( QingflowApiError( category="config", message="task field update is blocked because the current node does not expose a usable update schema", details={ "error_code": "TASK_UPDATE_SCHEMA_BLOCKED", "update_schema": update_schema, }, ) ) index = schema_state["index"] preflight = self._record_tools._build_record_write_preflight( profile=profile, context=context, operation="update", app_key=app_key, apply_id=record_id, answers=[], fields=fields, force_refresh_form=False, view_id=None, list_type=None, view_key=None, view_name=None, existing_answers_override=current_answers, field_index_override=index, ) effective_editable_ids = set(schema_state["effective_editable_question_ids"]) scoped_field_errors = self._task_scope_field_errors( normalized_answers=preflight.get("normalized_answers") or [], index=index, effective_editable_ids=effective_editable_ids, ) field_errors = list(preflight.get("field_errors") or []) field_errors.extend(scoped_field_errors) blockers = list(preflight.get("blockers") or []) if scoped_field_errors: blockers.append("payload writes fields that are not editable on the current task node") confirmation_requests = list(preflight.get("confirmation_requests") or []) if field_errors or confirmation_requests or blockers: raise_tool_error( QingflowApiError( category="config", message="task field update preflight was blocked", details={ "error_code": "TASK_FIELD_PLAN_BLOCKED", "blockers": blockers, "field_errors": field_errors, "confirmation_requests": confirmation_requests, "update_schema": update_schema, "recommended_next_actions": preflight.get("recommended_next_actions") or [], }, ) ) normalized_answers = [item for item in (preflight.get("normalized_answers") or []) if isinstance(item, dict)] merged_answers = self._record_tools._merge_record_answers(current_answers, normalized_answers) return { "update_schema": update_schema, "normalized_answers": normalized_answers, "merged_answers": merged_answers, } def _task_scope_field_errors( self, *, normalized_answers: list[dict[str, Any]], index: Any, effective_editable_ids: set[int], ) -> list[dict[str, Any]]: """执行内部辅助逻辑。""" if index is None: return [] field_errors: list[dict[str, Any]] = [] for answer in normalized_answers: que_id = _coerce_count(answer.get("queId")) if que_id is None or que_id <= 0: continue field = index.by_id.get(str(que_id)) field_payload = _field_ref_payload(field) if field is not None else {"que_id": que_id} if que_id not in effective_editable_ids: field_errors.append( { "location": field.que_title if field is not None else str(que_id), "message": "field is not editable on the current task node", "error_code": "TASK_FIELD_NOT_EDITABLE", "field": field_payload, } ) continue if field is None or field.que_type not in SUBTABLE_QUE_TYPES: continue table_values = answer.get("tableValues") if isinstance(answer.get("tableValues"), list) else [] subtable_index = self._record_tools._subtable_field_index_optional(field) for row_ordinal, row in enumerate(table_values, start=1): row_cells = [item for item in row if isinstance(item, dict)] if isinstance(row, list) else [] for cell in row_cells: cell_que_id = _coerce_count(cell.get("queId")) if cell_que_id is None or cell_que_id <= 0 or cell_que_id in effective_editable_ids: continue subfield = subtable_index.by_id.get(str(cell_que_id)) if subtable_index is not None else None field_errors.append( { "location": f"{field.que_title}[{row_ordinal}].{subfield.que_title if subfield is not None else cell_que_id}", "message": "subtable field is not editable on the current task node", "error_code": "TASK_FIELD_NOT_EDITABLE", "field": _field_ref_payload(subfield) if subfield is not None else {"que_id": cell_que_id}, } ) return field_errors def _task_save_only( self, *, profile: str, app_key: str, record_id: int, workflow_node_id: int, apply_answers: list[dict[str, Any]], ) -> dict[str, Any]: """执行内部辅助逻辑。""" def runner(session_profile, context): result = self.backend.request( "POST", context, f"/app/{app_key}/apply/{record_id}", json_body={"role": 3, "auditNodeId": workflow_node_id, "answers": apply_answers}, ) return { "profile": profile, "ws_id": session_profile.selected_ws_id, "app_key": app_key, "apply_id": record_id, "result": result, "request_route": self._request_route_payload(context), } return self._run(profile, runner, tool_name="任务仅保存") def _build_visibility(self, node_info: dict[str, Any], detail: dict[str, Any]) -> dict[str, bool]: """执行内部辅助逻辑。""" return { "comment_visible": self._coerce_bool(node_info.get("commentStatus")), "audit_record_visible": self._coerce_bool(node_info.get("auditRecordVisible")), "workflow_future_visible": self._coerce_bool(node_info.get("beingWorkflowNodeFutureListVisible")), "qrobot_record_visible": self._coerce_bool(node_info.get("qrobotRecordBeingVisible")), "associated_report_visible": self._resolve_associated_report_visible(node_info, detail), } def _resolve_associated_report_visible(self, node_info: dict[str, Any], detail: dict[str, Any]) -> bool: """执行内部辅助逻辑。""" node_visible = node_info.get("asosChartVisible") if node_visible is not None: return self._coerce_bool(node_visible) return self._coerce_bool(detail.get("viewAsosChartVisible")) def _normalize_associated_report(self, raw: dict[str, Any]) -> dict[str, Any]: """执行内部辅助逻辑。""" graph_type = str(raw.get("graphType") or "").strip().lower() source_type = str(raw.get("sourceType") or "").strip().lower() return { "report_id": raw.get("id"), "chart_key": raw.get("chartKey"), "chart_name": raw.get("chartName"), "graph_type": "view" if graph_type.endswith("view") or graph_type == "view" else "chart", "source_type": source_type or "qingflow", "target_app_key": raw.get("appKey"), "target_app_name": raw.get("formTitle"), "match_rules": raw.get("matchRules") or [], "raw": dict(raw), } def _rollback_candidate_items(self, payload: Any) -> list[dict[str, Any]]: """执行内部辅助逻辑。""" if isinstance(payload, dict): revert_nodes = payload.get("revertNodes") if isinstance(revert_nodes, list): return [item for item in revert_nodes if isinstance(item, dict)] return [item for item in _approval_page_items(payload) if isinstance(item, dict)] def _filter_transfer_members(self, items: Any, *, current_uid: int | None) -> list[dict[str, Any]]: """执行内部辅助逻辑。""" if not isinstance(items, list): return [] filtered: list[dict[str, Any]] = [] for item in items: if not isinstance(item, dict): continue uid = _coerce_count(item.get("uid") or item.get("userId") or item.get("memberId") or item.get("id")) if current_uid is not None and uid == current_uid: continue filtered.append(item) return filtered def _transfer_candidate_items( self, context: BackendRequestContext, *, app_key: str, record_id: int, workflow_node_id: int, current_uid: int | None, ) -> tuple[list[dict[str, Any]], list[JSONObject], JSONObject]: page_size = 100 max_pages = 100 page_num = 1 fetched_pages = 0 fetched_raw_count = 0 page_amount: int | None = None reported_total: int | None = None items: list[dict[str, Any]] = [] seen_member_keys: set[str] = set() warnings: list[JSONObject] = [] while page_num <= max_pages: try: result = self.backend.request( "GET", context, f"/app/{app_key}/apply/{record_id}/transfer/member", params={"pageNum": page_num, "pageSize": page_size, "auditNodeId": workflow_node_id}, ) except QingflowApiError as exc: if not _is_task_optional_read_error(exc): raise warnings.append( { "code": "TASK_TRANSFER_CANDIDATES_UNAVAILABLE", "message": "Transfer candidates are not readable in this permission context; task detail remains available.", "backend_code": exc.backend_code, "request_id": exc.request_id, "http_status": exc.http_status, } ) return items, warnings, { "loaded": False, "page_size": page_size, "fetched_pages": fetched_pages, "reported_total": reported_total, "page_amount": page_amount, "truncated": False, } fetched_pages += 1 raw_items = _approval_page_items(result) fetched_raw_count += len(raw_items) if page_amount is None: page_amount = _coerce_count(_approval_page_amount(result)) if reported_total is None: reported_total = _coerce_count(_approval_page_total(result)) for item in self._filter_transfer_members(raw_items, current_uid=current_uid): member_key = self._transfer_member_dedupe_key(item) if member_key in seen_member_keys: continue seen_member_keys.add(member_key) items.append(item) if not raw_items: break if page_amount is not None and page_num >= page_amount: break if reported_total is not None and fetched_raw_count >= reported_total: break page_num += 1 truncated = page_num > max_pages if truncated: warnings.append( { "code": "TRANSFER_CANDIDATES_TRUNCATED", "message": "transfer candidates reached the MCP safety page cap; returned candidates may be incomplete.", "max_pages": max_pages, "page_size": page_size, } ) pagination: JSONObject = { "loaded": True, "page_size": page_size, "fetched_pages": fetched_pages, "reported_total": reported_total, "page_amount": page_amount, "truncated": truncated, } return items, warnings, pagination def _transfer_member_dedupe_key(self, item: dict[str, Any]) -> str: uid = item.get("uid") or item.get("userId") or item.get("memberId") or item.get("id") if uid not in (None, ""): return f"uid:{uid}" return json.dumps(item, ensure_ascii=False, sort_keys=True, default=str) def _find_associated_report(self, task_context: dict[str, Any], report_id: int) -> dict[str, Any] | None: """执行内部辅助逻辑。""" associated_reports = ((task_context.get("associated_reports") or {}).get("items") or []) for item in associated_reports: if isinstance(item, dict) and item.get("report_id") == report_id: return item return None def _build_association_query(self, asos_chart: dict[str, Any], answers: list[dict[str, Any]]) -> dict[str, Any]: """执行内部辅助逻辑。""" key_que_ids = self._collect_match_rule_question_ids(asos_chart.get("matchRules") or []) key_values: list[dict[str, Any]] = [] for answer in answers: if not isinstance(answer, dict): continue answer_que_id = answer.get("queId") if isinstance(answer_que_id, int) and answer_que_id in key_que_ids: extracted_values = self._extract_answer_values(answer) key_values.append({"keyQueId": answer_que_id, "values": extracted_values or None}) table_values = answer.get("tableValues") if not isinstance(table_values, list): continue for idx, row in enumerate(table_values, start=1): if not isinstance(row, list): continue for sub_answer in row: if not isinstance(sub_answer, dict): continue sub_que_id = sub_answer.get("queId") if isinstance(sub_que_id, int) and sub_que_id in key_que_ids: extracted_values = self._extract_answer_values(sub_answer) key_values.append( { "keyQueId": sub_que_id, "ordinal": idx, "values": extracted_values or None, } ) return { "asosChart": self._sanitize_associated_chart(asos_chart), "keyQueValues": key_values, } def _collect_match_rule_question_ids(self, match_rules: Any) -> set[int]: """执行内部辅助逻辑。""" question_ids: set[int] = set() def visit(node: Any) -> None: if isinstance(node, list): for item in node: visit(item) return if not isinstance(node, dict): return for key in ("queId", "judgeQueId"): value = node.get(key) if isinstance(value, int): question_ids.add(value) for value in node.values(): if isinstance(value, (list, dict)): visit(value) visit(match_rules) return question_ids def _extract_answer_values(self, answer: dict[str, Any]) -> list[str]: """执行内部辅助逻辑。""" values = answer.get("values") if not isinstance(values, list): return [] normalized: list[str] = [] for item in values: if item is None: continue if isinstance(item, (str, int, float, bool)): normalized.append(str(item)) continue if not isinstance(item, dict): continue for key in ("value", "id", "uid", "userId", "applyId", "phone", "email", "name", "title", "label"): value = item.get(key) if value not in (None, ""): normalized.append(str(value)) break deduped: list[str] = [] seen: set[str] = set() for item in normalized: if item in seen: continue deduped.append(item) seen.add(item) return deduped def _sanitize_associated_chart(self, asos_chart: dict[str, Any]) -> dict[str, Any]: """执行内部辅助逻辑。""" return { "id": asos_chart.get("id"), "appKey": asos_chart.get("appKey"), "formTitle": asos_chart.get("formTitle"), "chartKey": asos_chart.get("chartKey"), "chartName": asos_chart.get("chartName"), "chartType": asos_chart.get("chartType"), "matchRules": asos_chart.get("matchRules") or [], "sourceType": asos_chart.get("sourceType"), "graphType": asos_chart.get("graphType"), "viewType": asos_chart.get("viewType"), } def _qingflow_chart_uses_apply_filter(self, context: BackendRequestContext, chart_key: str) -> bool: """执行内部辅助逻辑。""" if not chart_key: return False try: auth = self.backend.request( "GET", context, f"/chart/{chart_key}/auth", ) except QingflowApiError as error: if not _is_task_optional_read_error(error): raise return False if not isinstance(auth, dict): return False return self._coerce_bool(auth.get("detailedViewStatus")) and auth.get("lastViewType") == 1 def _normalize_chart_result(self, payload: Any) -> dict[str, Any]: """执行内部辅助逻辑。""" if isinstance(payload, dict): rows = payload.get("rows") if not isinstance(rows, list): rows = payload.get("list") if isinstance(payload.get("list"), list) else [] series = payload.get("series") if not isinstance(series, list): series = payload.get("xAxis") if isinstance(payload.get("xAxis"), list) else [] metrics = payload.get("metrics") if not isinstance(metrics, list): metrics = payload.get("yAxis") if isinstance(payload.get("yAxis"), list) else [] summary = payload.get("summary") if isinstance(payload.get("summary"), dict) else {} return { "summary": summary, "rows": rows or [], "series": series or [], "metrics": metrics or [], } if isinstance(payload, list): return {"summary": {}, "rows": payload, "series": [], "metrics": []} return {"summary": {}, "rows": [], "series": [], "metrics": []} def _normalize_workflow_logs(self, payload: Any) -> list[dict[str, Any]]: """执行内部辅助逻辑。""" if isinstance(payload, dict): page = payload.get("list") if isinstance(payload.get("list"), list) else payload.get("rows") if not isinstance(page, list): nested = payload.get("data") if isinstance(nested, dict): page = nested.get("list") if isinstance(nested.get("list"), list) else nested.get("rows") if not isinstance(page, list): page = [] elif isinstance(payload, list): page = payload else: page = [] items: list[dict[str, Any]] = [] for node_record in page: if not isinstance(node_record, dict): continue node_id = node_record.get("nodeId") node_name = node_record.get("nodeName") operation_record_list = node_record.get("operationRecordList") if not isinstance(operation_record_list, list): operation_record_list = [] for operation in operation_record_list: if not isinstance(operation, dict): continue detail = self._first_nested_operation_detail(operation) items.append( { "log_id": operation.get("workflowNodeOperationRecordId") or node_record.get("workflowNodeProcessRecordId"), "node_id": node_id, "node_name": node_name, "operator": operation.get("operator"), "operation": operation.get("operationType"), "operation_result": detail, "operation_time": operation.get("operationTime"), "remark": self._extract_remark(detail), "signature_url": self._extract_signature_url(detail), "attachments": self._extract_attachments(detail), "qrobot_related": any( operation.get(key) for key in ("qRobotAdd", "qRobotUpdate", "qRobotSMS", "qRobotMail", "webhook") ), } ) return items def _workflow_log_digest(self, items: list[dict[str, Any]]) -> str | None: """执行内部辅助逻辑。""" if not items: return None try: return json.dumps(items, ensure_ascii=False, sort_keys=True, default=str) except TypeError: return str(items) def _first_nested_operation_detail(self, operation: dict[str, Any]) -> Any: """执行内部辅助逻辑。""" for key in ("approval", "filling", "cc", "applicant", "qRobotAdd", "qRobotUpdate", "webhook", "qRobotSMS", "qRobotMail"): value = operation.get(key) if value is not None: return value return None def _extract_remark(self, detail: Any) -> Any: """执行内部辅助逻辑。""" if not isinstance(detail, dict): return None for key in ("remark", "feedback", "comment", "content"): value = detail.get(key) if value not in (None, ""): return value return None def _extract_signature_url(self, detail: Any) -> Any: """执行内部辅助逻辑。""" if not isinstance(detail, dict): return None for key in ("signatureUrl", "handSignImageUrl"): value = detail.get(key) if value not in (None, ""): return value return None def _extract_attachments(self, detail: Any) -> Any: """执行内部辅助逻辑。""" if not isinstance(detail, dict): return [] for key in ("attachments", "files", "uploadFiles"): value = detail.get(key) if isinstance(value, list): return value return [] def _extract_audit_feedback(self, payload: dict[str, Any]) -> str | None: """执行内部辅助逻辑。""" for key in ("audit_feedback", "auditFeedback"): value = payload.get(key) if isinstance(value, str) and value.strip(): return value.strip() return None def _extract_positive_int(self, payload: dict[str, Any], key: str, *, aliases: tuple[str, ...] = ()) -> int: """执行内部辅助逻辑。""" candidates = (key, *aliases) value: Any = None for candidate in candidates: if candidate in payload: value = payload.get(candidate) break if not isinstance(value, int) or value <= 0: names = ", ".join(candidates) raise_tool_error(QingflowApiError.config_error(f"one of [{names}] must be a positive integer")) return value def _require_app_record_and_node(self, app_key: str, record_id: int, workflow_node_id: int) -> None: """执行内部辅助逻辑。""" if not app_key: raise_tool_error(QingflowApiError.config_error("app_key is required")) if record_id <= 0: raise_tool_error(QingflowApiError.config_error("record_id must be positive")) if workflow_node_id <= 0: raise_tool_error(QingflowApiError.config_error("workflow_node_id must be positive")) def _coerce_bool(self, value: Any) -> bool: """执行内部辅助逻辑。""" if isinstance(value, bool): return value if isinstance(value, int): return value != 0 if isinstance(value, str): return value.strip().lower() in {"1", "true", "yes", "y", "show", "visible", "enabled"} return bool(value) def _request_route_payload(self, context: BackendRequestContext) -> dict[str, Any]: """执行内部辅助逻辑。""" describe_route = getattr(self.backend, "describe_route", None) if callable(describe_route): payload = describe_route(context) if isinstance(payload, dict): return payload return { "base_url": context.base_url, "qf_version": context.qf_version, "qf_version_source": context.qf_version_source or ("context" if context.qf_version else "unknown"), } def _is_optional_task_audit_info_error(error: QingflowApiError) -> bool: """Whether task auditInfo may be replaced by current task detail.""" if is_auth_like_error(error): return False return _is_task_permission_context_error(error) or error.http_status == 404 def _task_box_record_list_type(task_box: str | None) -> int: normalized = str(task_box or "todo").strip().lower() if normalized == "initiated": return 14 if normalized == "done": return 2 if normalized == "cc": return 12 return 1 def _task_box_record_role(task_box: str | None) -> int: normalized = str(task_box or "todo").strip().lower() if normalized == "initiated": return 2 return 3 def _is_optional_task_box_locator_error(error: QingflowApiError) -> bool: """Whether one task box may be skipped while resolving a task_id.""" if is_auth_like_error(error): return False return _is_task_permission_context_error(error) or error.http_status == 404 def _is_task_optional_read_error(error: QingflowApiError) -> bool: if is_auth_like_error(error): return False backend_code = _task_backend_code(error) return backend_code in {40002, 40027, 404} or error.http_status == 404 def _is_task_permission_context_error(error: QingflowApiError) -> bool: if is_auth_like_error(error): return False return _task_backend_code(error) in {40002, 40027} def _is_permission_context_error_payload(payload: dict[str, Any]) -> bool: if str(payload.get("category") or "").strip().lower() == "auth": return False if message_looks_like_invalid_token(payload.get("message")): return False if backend_code_value_int(payload.get("http_status")) == 401: return False return backend_code_value_int(payload.get("backend_code")) in {40002, 40027} def _task_backend_code(error: QingflowApiError) -> int | None: return backend_code_int(error)