from __future__ import annotations import re from copy import deepcopy from typing import Any from ..card_details import CardBusinessDataProjector, CardDataNotFound from ..card_details.common import find_detail from ..inspector_projection import project_activity_detail, project_event_detail from ..module_audit.log_context import locate_log_module from ..retrieval_detail_projection import ( QUERY_TOOL_NAMES, RETRIEVAL_RESULT_COLLECTION_KEYS, ) from .comparison import align_business_projection, business_only_projection from .resolver import SourceResolver, pointer_value from .schema import SourceBinding, unresolved_binding class InspectorViewNotFound(LookupError): pass class InspectorWorkbenchProjector: """Application service that makes all three Inspector columns together.""" def __init__(self) -> None: self._card_projector = CardBusinessDataProjector() def related_event_ids( self, detail_ref: str, *, bundle: dict[str, Any], view: dict[str, Any] ) -> list[int]: ref = str(detail_ref or "") ids: set[int] = set() if ref.startswith("event:"): value = _suffix_int(ref) if value: ids.add(value) node = find_detail(view, ref) _collect_event_ids(node, ids) if ref == "run:objective": _collect_event_ids(view.get("objective"), ids) round_match = re.fullmatch(r"round:(\d+):(goal|plan|result|evaluation)", ref) if round_match: round_index = int(round_match.group(1)) round_ = next( ( item for item in view.get("rounds") or [] if _integer(item.get("roundIndex")) == round_index ), {}, ) role = round_match.group(2) target = { "goal": round_.get("goal"), "plan": (round_.get("plan") or {}).get("node"), "result": round_.get("overallEvaluation"), "evaluation": round_.get("overallEvaluation"), }.get(role) _collect_event_ids(target, ids) event_names = { "goal": {"begin_round"}, "plan": {"record_multipath_plan"}, }.get(role, set()) for event in bundle.get("events") or []: if ( event.get("event_name") in event_names and ( role == "goal" or _integer(event.get("round_index")) == round_index ) ): ids.add(_integer(event.get("id")) or 0) branch_match = re.fullmatch( r"round:(\d+):branch:(\d+):(task|retrieval|output)", ref ) if branch_match: round_index, branch_id = int(branch_match.group(1)), int(branch_match.group(2)) branch = _view_branch(view, round_index, branch_id) _collect_event_ids(branch, ids) if ref.startswith("creative:"): seed_id = _suffix_int(ref) seed = _bundle_event(bundle, seed_id) round_index = _integer((seed or {}).get("round_index")) branch_id = _integer((seed or {}).get("branch_id")) implementers = { _integer(event.get("id")) for event in bundle.get("events") or [] if event.get("event_name") == "script_implementer" and _integer(event.get("round_index")) == round_index and _integer(event.get("branch_id")) == branch_id } for event in bundle.get("events") or []: if _integer(event.get("round_index")) != round_index or _integer(event.get("branch_id")) != branch_id: continue event_id = _integer(event.get("id")) if event_id in implementers or _integer(event.get("scope_event_id")) in implementers: ids.add(event_id or 0) if ref.startswith("multipath-decision:"): decision_id = _suffix_int(ref) decision = next( ( item for item in bundle.get("multipathDecisions") or [] if _integer(item.get("id")) == decision_id ), {}, ) round_index = _integer(decision.get("round_index")) for event in bundle.get("events") or []: if ( event.get("event_name") == "script_multipath_evaluator" and _integer(event.get("round_index")) == round_index ): ids.add(_integer(event.get("id")) or 0) if ref.startswith("retrieval-agent:"): agent_id = _suffix_int(ref) for event in bundle.get("events") or []: if ( _integer(event.get("id")) == agent_id or _integer(event.get("parent_event_id")) == agent_id or _integer(event.get("scope_event_id")) == agent_id ): ids.add(_integer(event.get("id")) or 0) return sorted(value for value in ids if value > 0) def project( self, detail_ref: str, *, script_build_id: int, bundle: dict[str, Any], view: dict[str, Any], event_details: dict[int, dict[str, Any]], log_contents: dict[int, str] | None = None, ) -> dict[str, Any]: ref = str(detail_ref or "").strip() def event_loader(event_id: int) -> dict[str, Any]: event = event_details.get(int(event_id)) if event is None: raise KeyError(event_id) return event business = self._business_detail( ref, script_build_id=script_build_id, bundle=bundle, view=view, event_details=event_details, ) try: card_data = self._card_projector.project( ref, bundle=bundle, view=view, load_event=event_loader, ) except (CardDataNotFound, KeyError): card_data = _fallback_card_data(ref, business) resolver = SourceResolver(bundle=bundle, event_details=event_details) source_modules = self._modules(ref, business, card_data, resolver, event_details, bundle) if not source_modules: raise InspectorViewNotFound(f"未找到活动 {ref}") alignment_kind = ( "retrieval-event" if business.get("detailKind") == "retrieval-event" else str(card_data.get("cardKind") or _fallback_card_kind(ref)) ) modules = align_business_projection( detail_ref=ref, card_kind=alignment_kind, business=business, source_modules=source_modules, resolver=resolver, bundle=bundle, ) if not modules: raise InspectorViewNotFound(f"活动 {ref} 没有可展示的业务详情") if log_contents: log_refs: dict[int, str] = {} for event_id, event in event_details.items(): log_module = locate_log_module(log_contents.get(event_id) or "", event) if log_module: log_refs[event_id] = resolver.add_log_anchor( event_id, log_module, event_msg_id=str(event.get("msg_id") or ""), ) for module in modules: runtime_refs = list(module.get("runtimeRefs") or []) for event_id, log_ref in log_refs.items(): if f"event:{event_id}" in runtime_refs and log_ref not in runtime_refs: runtime_refs.append(log_ref) module["runtimeRefs"] = runtime_refs completeness = str(card_data.get("completeness") or "partial") if any( binding.get("resolution") != "resolved" for binding in _all_bindings(modules) ): completeness = "partial" if completeness != "missing" else "missing" notices = [ *[item for item in card_data.get("notices") or [] if isinstance(item, dict)], *_business_notices(business), ] used_source_ids = _used_source_ids(modules) return { "schemaVersion": "inspector-source-v2", "detailRef": ref, "cardKind": card_data.get("cardKind") or _fallback_card_kind(ref), "completeness": completeness, "businessProjection": business_only_projection(business), "modules": modules, "sources": { source_id: source for source_id, source in resolver.sources.items() if source_id in used_source_ids }, "notices": _unique_notices(notices), } def _business_detail( self, detail_ref: str, *, script_build_id: int, bundle: dict[str, Any], view: dict[str, Any], event_details: dict[int, dict[str, Any]], ) -> dict[str, Any]: if detail_ref.startswith("event:"): event_id = _suffix_int(detail_ref) event = event_details.get(event_id or -1) if event is None: raise InspectorViewNotFound(f"未找到运行事件 {event_id}") try: return project_activity_detail( script_build_id, detail_ref, view, bundle, event_detail=event, ) except KeyError: return project_event_detail( script_build_id, event, run_status=((view.get("header") or {}).get("status")), ) event_detail = None if detail_ref.startswith("retrieval-agent:"): run = find_detail(view, detail_ref) or {} event_detail = event_details.get(_integer(run.get("eventId")) or -1) try: return project_activity_detail( script_build_id, detail_ref, view, bundle, event_detail=event_detail, ) except KeyError as exc: raise InspectorViewNotFound(f"未找到活动 {detail_ref}") from exc def _modules( self, detail_ref: str, business: dict[str, Any], card_data: dict[str, Any], resolver: SourceResolver, event_details: dict[int, dict[str, Any]], bundle: dict[str, Any], ) -> list[dict[str, Any]]: if detail_ref.startswith("retrieval-direct:"): return _direct_group_modules(card_data, resolver, event_details) if detail_ref.startswith("retrieval-agent:"): return _retrieval_agent_modules(card_data, resolver, event_details) special = self._event_modules(detail_ref, business, resolver, event_details) if special is not None: return special business_values = _business_modules(business) planned = [ item for item in (card_data.get("displayUse") or {}).get("businessModules") or [] if isinstance(item, dict) ] fields = { str(item.get("id")): item for item in [ *(card_data.get("businessInputs") or []), *(card_data.get("businessOutputs") or []), ] if isinstance(item, dict) and item.get("id") } if card_data.get("cardKind") == "implementation-task" and not any( str((field.get("source") or {}).get("kind") or "") == "runtime-event" for field in fields.values() ): return _historical_implementation_task_modules(fields, resolver) modules: list[dict[str, Any]] = [] consumed: set[str] = set() for index, plan in enumerate(planned, 1): module_id = str(plan.get("id") or f"module-{index}") title = str(plan.get("title") or f"业务模块 {index}") business_module = _match_business_module(business_values, module_id, title) if business_module: consumed.add(str(business_module.get("id"))) source_ids = [str(value) for value in plan.get("sourceIds") or []] bindings = [ resolver.binding_for_field( fields[source_id], binding_id=f"{module_id}:source:{position}", ) for position, source_id in enumerate(source_ids, 1) if source_id in fields ] if card_data.get("cardKind") == "implementation-task" and business_module: task_field = next( ( fields[source_id] for source_id in source_ids if source_id in fields and str((fields[source_id].get("source") or {}).get("kind")) == "runtime-event" and str((fields[source_id].get("source") or {}).get("fieldPath")) == "input.content.task" ), None, ) task_event_id = _suffix_int(str((task_field or {}).get("source", {}).get("ref") or "")) if task_event_id and _module_value(business_module): bindings = [ resolver.text_span_binding( task_event_id, binding_id=f"{module_id}:task-span", path="input.content.task", exact_text=_module_value(business_module), role="input", ), *[ binding for binding in bindings if binding.get("sourceId") != f"event:{task_event_id}" ], ] aggregate = _aggregate_binding( str(card_data.get("cardKind") or ""), module_id, source_ids, fields, resolver, bundle, ) if aggregate is not None: bindings = [aggregate] if module_id == "process" and not bindings: bindings = [ resolver.binding_for_event( event_id, binding_id=f"{module_id}:event:{event_id}", role="process", ) for event_id in sorted(event_details) ] if card_data.get("cardKind") == "multipath-decision" and module_id == "outcomes": for binding in bindings: source_record = resolver.sources.get(str(binding.get("sourceId") or "")) or {} raw_record = source_record.get("rawRecord") branch_status = str((raw_record or {}).get("status") or "").lower() if isinstance(raw_record, dict) else "" if branch_status in {"merged", "adopted", "accepted", "selected"}: binding["evidence"]["adoption"] = "explicitly-adopted" binding["evidence"]["confidence"] = "exact" elif branch_status in {"discarded", "rejected", "not_adopted"}: binding["evidence"]["adoption"] = "explicitly-rejected" binding["evidence"]["confidence"] = "exact" if not bindings: binding = unresolved_binding( f"{module_id}:unresolved", reason=f"模块「{title}」没有保存可安全定位的原始记录", code="binding-map-missing", ) resolver.add_unresolved_source(binding) bindings = [binding] value = _module_value(business_module) if value in (None, "", [], {}) and source_ids: values = [fields[source_id].get("value") for source_id in source_ids if source_id in fields] value = values[0] if len(values) == 1 else values _align_transforms_with_business(value, bindings, resolver, module_id) module = { "id": module_id, "title": title, "presentation": _presentation(business_module, title), "business": {"value": value}, "bindings": bindings, "runtimeRefs": _runtime_refs(bindings), "defaultOpen": index <= 3, } if business_module and business_module.get("items"): business_items = _normalize_items(business_module.get("items") or []) item_rows = [] child_bindings: list[SourceBinding] = [] for item_index, item in enumerate(business_items, 1): item_bindings = ( [_multipath_comparison_binding( item, item_index, bindings, resolver, bundle, )] if card_data.get("cardKind") == "multipath-decision" and module_id == "comparison" else _item_bindings( module_id, item_index, item, len(business_items), bindings, resolver, ) ) child_bindings.extend(item_bindings) item_rows.append( { "id": f"{module_id}:item:{item_index}", "label": item.get("label"), "value": item.get("value"), "bindings": item_bindings, } ) module["business"] = {"items": item_rows} module["bindings"] = _unique_bindings(child_bindings) module["runtimeRefs"] = _runtime_refs(module["bindings"]) elif business_module and business_module.get("steps"): step_rows = [] step_bindings: list[SourceBinding] = [] for step_index, step in enumerate(business_module.get("steps") or [], 1): if not isinstance(step, dict): continue step_event_id = _suffix_int(str(step.get("eventRef") or "")) if step_event_id and step_event_id in event_details: step_binding = resolver.binding_for_event( step_event_id, binding_id=f"{module_id}:step:{step_index}", path="", role="process", ) step_binding["transform"] = { "kind": "parsed", "operation": "creative-step-v1", "outputKey": str(step_index), } step_binding["evidence"]["confidence"] = "deterministic" else: step_binding = unresolved_binding( f"{module_id}:step:{step_index}:unresolved", reason=f"创作步骤 {step_index} 没有保存可安全关联的 Event", source_id=(f"event:{step_event_id}" if step_event_id else None), resolution="unsafe", code="unsafe-association", ) if step_binding["sourceId"] not in resolver.sources: resolver.add_unresolved_source(step_binding) step_bindings.append(step_binding) step_rows.append({ "id": f"{module_id}:step:{step_index}", "label": step.get("action") or f"步骤 {step_index}", "value": step, "bindings": [step_binding], }) module["business"] = {"items": step_rows} module["bindings"] = step_bindings module["runtimeRefs"] = _runtime_refs(step_bindings) modules.append(module) # The formal Inspector exposes only modules declared by the unified # card contract. Legacy business blocks are an input adapter, not a # second set of UI modules; appending unmatched blocks here created # duplicates such as Goal + “本轮目标构成” with no trustworthy binding. return modules def _event_modules( self, detail_ref: str, business: dict[str, Any], resolver: SourceResolver, event_details: dict[int, dict[str, Any]], ) -> list[dict[str, Any]] | None: if not detail_ref.startswith("event:"): return None event_id = _suffix_int(detail_ref) event = event_details.get(event_id or -1) if event is None: return None if business.get("detailKind") == "retrieval-event": return _retrieval_event_modules(business, event, resolver) if event.get("event_name") == "script_implementer": input_side = event.get("input") if isinstance(event.get("input"), dict) else {} content = input_side.get("content") if isinstance(input_side, dict) else None task = content.get("task") if isinstance(content, dict) else content sections = { str(item.get("id")): item for item in business.get("businessSections") or [] if isinstance(item, dict) } action_binding = resolver.binding_for_event( event_id or 0, binding_id="action:event", path="event_name", role="process", ) action_binding["transform"] = { "kind": "parsed", "operation": "implementation-activity-label-v1", "outputKey": "action", } action_binding["evidence"]["confidence"] = "deterministic" request_binding = resolver.binding_for_event( event_id or 0, binding_id="request:event", path="input.content.task", role="input", availability="direct-read", ) request_binding["transform"] = { "kind": "parsed", "operation": "implementation-task-readable-sections-v1", "outputKey": "request", } request_binding["evidence"]["confidence"] = "deterministic" result_value = (sections.get("result") or {}).get("content") result_binding = resolver.text_span_binding( event_id or 0, binding_id="result:event", path="output.content.summary", exact_text=result_value, role="output", ) return [ { "id": "action", "title": "做了什么", "presentation": "text", "business": {"value": (sections.get("action") or {}).get("content")}, "bindings": [action_binding], "runtimeRefs": [detail_ref], "defaultOpen": True, }, { "id": "request", "title": "处理内容", "presentation": "text", "business": {"value": (sections.get("request") or {}).get("content")}, "bindings": [request_binding], "runtimeRefs": [detail_ref], "defaultOpen": True, }, { "id": "result", "title": "得到什么", "presentation": "text", "business": {"value": result_value}, "bindings": [result_binding], "runtimeRefs": [detail_ref], "defaultOpen": True, }, ] values = _business_modules(business) planning_paths = { "summary": "input.content.thought_summary", "thought": "input.content.thought", "plan": "input.content.plan", "action": "input.content.action", } modules: list[dict[str, Any]] = [] is_planning = ( event.get("event_name") == "think_and_plan" and event.get("agent_role") == "main" and _integer(event.get("agent_depth")) == 0 ) is_review = event.get("event_name") in {"script_multipath_evaluator", "script_evaluator"} if not is_planning and not is_review: return _generic_event_modules( detail_ref, business, event, resolver, event_id or 0 ) for index, item in enumerate(values, 1): module_id = str(item.get("id") or f"event-module-{index}") title = str(item.get("title") or f"运行模块 {index}") value = _module_value(item) input_title = title in {"评审范围", "评审标准", "评审对象", "标准"} if is_planning and module_id in planning_paths: binding = resolver.binding_for_event( event_id or 0, binding_id=f"{module_id}:event:{event_id}", path=planning_paths[module_id], role="input", availability="direct-read", ) elif title == "记录说明" and not event.get("output"): binding = resolver.binding_for_event( event_id or 0, binding_id=f"{module_id}:event:{event_id}", role="status", availability="produced-by-run", ) binding["transform"] = { "kind": "calculated", "operation": "runtime-missing-decision-notice-v1", } binding["evidence"]["confidence"] = "deterministic" else: side = "input" if input_title else "output" binding = resolver.text_span_binding( event_id or 0, binding_id=f"{module_id}:event:{event_id}", path=_event_text_path(event, side), exact_text=value, role="input" if input_title else "output", ) business_payload: dict[str, Any] = {"value": value} bindings = [binding] if item.get("items"): rows = [] bindings = [] for item_index, child in enumerate(_normalize_items(item.get("items") or []), 1): child_bindings = [] for leaf_index, leaf in enumerate(_leaf_texts(child.get("value")), 1): child_bindings.append(resolver.text_span_binding( event_id or 0, binding_id=f"{module_id}:item:{item_index}:leaf:{leaf_index}", path=_event_text_path(event, "input" if input_title else "output"), exact_text=leaf, role="input" if input_title else "output", )) if not child_bindings: fallback = resolver.binding_for_event( event_id or 0, binding_id=f"{module_id}:item:{item_index}:record", path=_event_text_path(event, "input" if input_title else "output"), role="input" if input_title else "output", availability="direct-read" if input_title else "returned-to-agent", ) fallback["transform"] = { "kind": "parsed", "operation": "review-item-v1", "outputKey": str(item_index), } fallback["evidence"]["confidence"] = "deterministic" child_bindings = [fallback] bindings.extend(child_bindings) rows.append( { "id": f"{module_id}:item:{item_index}", "label": child.get("label"), "value": child.get("value"), "bindings": child_bindings, } ) business_payload = {"items": rows} modules.append( { "id": module_id, "title": title, "presentation": _presentation(item, title), "business": business_payload, "bindings": bindings, "runtimeRefs": [detail_ref], "defaultOpen": index <= 3, } ) return modules def _generic_event_modules( detail_ref: str, business: dict[str, Any], event: dict[str, Any], resolver: SourceResolver, event_id: int, ) -> list[dict[str, Any]] | None: """Bind the readable action/request/result projection to the real Event Body.""" sections = [ item for item in business.get("businessSections") or [] if isinstance(item, dict) ] if not sections: return None modules: list[dict[str, Any]] = [] for index, section in enumerate(sections, 1): module_id = str(section.get("id") or f"section-{index}") value = section.get("content") if module_id == "action": path, role, availability = "event_name", "process", "produced-by-run" operation = "runtime-activity-label-v1" elif module_id == "request": path, role, availability = _event_text_path(event, "input"), "input", "direct-read" operation = "runtime-request-readable-projection-v1" elif module_id == "result": path, role, availability = _event_text_path(event, "output"), "output", "returned-to-agent" operation = "runtime-result-readable-projection-v1" else: # The section is visibly derived from this Event, but there is no # card-specific field contract. Bind the complete Event instead of # falsely claiming that its original record is missing. path, role, availability = "", "process", "produced-by-run" operation = "runtime-business-section-v1" binding = resolver.binding_for_event( event_id, binding_id=f"{module_id}:event:{event_id}", path=path, role=role, availability=availability, ) binding["transform"] = { "kind": "parsed", "operation": operation, "outputKey": module_id, } binding["evidence"]["confidence"] = "deterministic" modules.append({ "id": module_id, "title": str(section.get("title") or f"业务模块 {index}"), "presentation": "text", "business": {"value": value}, "bindings": [binding], "runtimeRefs": [detail_ref], "defaultOpen": True, }) return modules def _retrieval_event_modules( business: dict[str, Any], event: dict[str, Any], resolver: SourceResolver ) -> list[dict[str, Any]]: event_id = _integer(event.get("id")) or 0 conditions = business.get("queryConditions") or [] condition_items = [] condition_bindings: list[SourceBinding] = [] for index, item in enumerate(conditions, 1): key = str(item.get("key") or index) binding = resolver.binding_for_event( event_id, binding_id=f"purpose:{key}", path=f"input.content.{key}", role="input", availability="direct-read", ) condition_bindings.append(binding) condition_items.append( { "id": f"purpose:item:{key}", "label": item.get("label"), "value": item.get("value"), "bindings": [binding], } ) if not condition_bindings: binding = resolver.binding_for_event( event_id, binding_id="purpose:event-input", path="input.content", role="input", availability="direct-read", ) condition_bindings = [binding] name = str(event.get("event_name") or "") result_key = RETRIEVAL_RESULT_COLLECTION_KEYS.get(name) root = "output.content" + (f".{result_key}" if result_key else "") root_value = pointer_value(event, f"/{root.replace('.', '/')}") result_items = [] result_bindings: list[SourceBinding] = [] for index, item in enumerate(business.get("items") or [], 1): item_path = ( f"{root}.{index - 1}" if isinstance(root_value, list) else root ) binding = resolver.binding_for_event( event_id, binding_id=f"result:item:{index}", path=item_path, role="output", availability="returned-to-agent", ) binding["transform"] = { "kind": "parsed", "operation": "retrieval-result-item-v1", "outputKey": str(index - 1), } binding["evidence"]["confidence"] = "deterministic" result_bindings.append(binding) result_items.append( { "id": f"result:item:{index}", "label": item.get("title"), "value": item, "bindings": [binding], } ) if not result_bindings: empty_result = resolver.binding_for_event( event_id, binding_id="result:event-output", path="output.content", role="output", availability="returned-to-agent", ) empty_result["transform"] = { "kind": "parsed", "operation": "retrieval-empty-or-failure-message-v1", "outputKey": "message", } empty_result["evidence"]["confidence"] = "deterministic" result_bindings = [empty_result] status_bindings = _event_status_bindings( resolver, event_id, prefix="status", include_output=True ) event_source = resolver.sources.get(f"event:{event_id}") if event_source is not None and str((business.get("outcome") or {}).get("state")) == "empty": event_source["resultState"] = "empty" return [ { "id": "purpose", "title": "目的 / 查询条件", "presentation": "items", "business": {"items": condition_items} if condition_items else {"value": "未记录调用输入"}, "bindings": condition_bindings, "runtimeRefs": [f"event:{event_id}"], "defaultOpen": True, }, { "id": "result", "title": "返回内容 / 结果列表", "presentation": "items", "business": {"items": result_items} if result_items else {"value": (business.get("outcome") or {}).get("message")}, "bindings": result_bindings, "runtimeRefs": [f"event:{event_id}"], "defaultOpen": True, }, { "id": "status", "title": "状态、耗时与完整性", "presentation": "notice", "business": {"value": {"outcome": business.get("outcome"), "completeness": business.get("completeness")}}, "bindings": status_bindings, "runtimeRefs": [f"event:{event_id}"], "defaultOpen": True, }, ] def _direct_group_modules( card_data: dict[str, Any], resolver: SourceResolver, event_details: dict[int, dict[str, Any]], ) -> list[dict[str, Any]]: events = [event_details[event_id] for event_id in sorted(event_details)] def rows(kind: str) -> tuple[list[dict[str, Any]], list[SourceBinding]]: values: list[dict[str, Any]] = [] bindings: list[SourceBinding] = [] for index, event in enumerate(events, 1): event_id = _integer(event.get("id")) or 0 if kind == "input": path, role, availability = "input.content", "input", "direct-read" label = str(event.get("event_name") or f"调用 {index}") value = ((event.get("input") or {}).get("content") if isinstance(event.get("input"), dict) else None) elif kind == "output": path, role, availability = "output.content", "output", "returned-to-agent" label = str(event.get("event_name") or f"调用 {index}") value = ((event.get("output") or {}).get("content") if isinstance(event.get("output"), dict) else None) else: path, role, availability = "status", "status", "produced-by-run" label = str(event.get("event_name") or f"调用 {index}") value = { "status": event.get("status"), "durationMs": event.get("duration_ms"), "error": event.get("error_message"), } item_bindings = ( _event_status_bindings(resolver, event_id, prefix=kind) if kind == "status" else [resolver.binding_for_event( event_id, binding_id=f"{kind}:event:{event_id}", path=path, role=role, availability=availability, )] ) bindings.extend(item_bindings) values.append( { "id": f"{kind}:item:{event_id}", "label": label, "value": value, "bindings": item_bindings, } ) return values, bindings input_rows, input_bindings = rows("input") output_rows, output_bindings = rows("output") status_rows, status_bindings = rows("status") failed_rows = [ row for row in status_rows if str((row.get("value") or {}).get("status") or "").lower() in {"failed", "failure", "error", "interrupted"} ] failed_ids = {row["id"].rsplit(":", 1)[-1] for row in failed_rows} failed_bindings = [ binding for binding in status_bindings if binding["id"].rsplit(":", 1)[-1] in failed_ids ] if not failed_rows: failed_bindings = [resolver.members_binding( { "id": "failures:none", "label": "失败调用数量", "value": 0, "relation": "persisted-output", }, binding_id="failures:none-calculated", members=[{ "id": f"failures:status:{event.get('id')}", "label": str(event.get("event_name") or event.get("id")), "value": event.get("status"), "source": { "kind": "runtime-event", "label": str(event.get("event_name") or f"Event {event.get('id')}"), "ref": f"event:{event.get('id')}", "fieldPath": "status", }, "relation": "persisted-output", "completeness": "complete", } for event in events], reducer="no-failed-tool-status-v1", role="status", )] return [ { "id": "purpose", "title": "读取目的", "presentation": "items", "business": {"items": input_rows}, "bindings": input_bindings, "runtimeRefs": _runtime_refs(input_bindings), "defaultOpen": True, }, { "id": "calls", "title": "逐次工具调用", "presentation": "items", "business": {"items": status_rows}, "bindings": status_bindings, "runtimeRefs": _runtime_refs(status_bindings), "defaultOpen": True, }, { "id": "results", "title": "实际返回的信息", "presentation": "items", "business": {"items": output_rows}, "bindings": output_bindings, "runtimeRefs": _runtime_refs(output_bindings), "defaultOpen": True, }, { "id": "failures", "title": "失败与缺失", "presentation": "notice", "business": {"items": failed_rows} if failed_rows else {"value": "未发现失败或中断的工具调用。"}, "bindings": failed_bindings, "runtimeRefs": _runtime_refs(failed_bindings), "defaultOpen": bool(failed_rows), }, ] def _retrieval_agent_modules( card_data: dict[str, Any], resolver: SourceResolver, event_details: dict[int, dict[str, Any]], ) -> list[dict[str, Any]]: events = [event_details[event_id] for event_id in sorted(event_details)] agent = next( (event for event in events if str(event.get("event_type") or "") == "agent_invoke"), events[0] if events else {}, ) agent_id = _integer(agent.get("id")) or 0 children = [event for event in events if _integer(event.get("id")) != agent_id] queries = [ event for event in children if str(event.get("event_name") or "") in QUERY_TOOL_NAMES ] task_binding = resolver.binding_for_event( agent_id, binding_id="task:agent-input", path="input.content.task", role="input", availability="direct-read", ) screening_binding = resolver.binding_for_event( agent_id, binding_id="screening:agent-output", path="output.content", role="output", availability="produced-by-run", ) task = ((agent.get("input") or {}).get("content") if isinstance(agent.get("input"), dict) else None) task = task.get("task") if isinstance(task, dict) else task screening = ((agent.get("output") or {}).get("content") if isinstance(agent.get("output"), dict) else None) query_rows: list[dict[str, Any]] = [] result_rows: list[dict[str, Any]] = [] status_rows: list[dict[str, Any]] = [] query_bindings: list[SourceBinding] = [] result_bindings: list[SourceBinding] = [] status_bindings: list[SourceBinding] = [] for index, event in enumerate(queries, 1): event_id = _integer(event.get("id")) or 0 label = str(event.get("event_name") or f"查询 {index}") input_binding = resolver.binding_for_event( event_id, binding_id=f"queries:event:{event_id}", path="input.content", role="input", availability="direct-read", ) output_binding = resolver.binding_for_event( event_id, binding_id=f"raw:event:{event_id}", path="output.content", role="output", availability="returned-to-agent", ) status_item_bindings = _event_status_bindings( resolver, event_id, prefix="failures" ) query_bindings.append(input_binding) result_bindings.append(output_binding) status_bindings.extend(status_item_bindings) query_rows.append({ "id": f"queries:item:{event_id}", "label": label, "value": ((event.get("input") or {}).get("content") if isinstance(event.get("input"), dict) else None), "bindings": [input_binding], }) result_rows.append({ "id": f"raw:item:{event_id}", "label": label, "value": ((event.get("output") or {}).get("content") if isinstance(event.get("output"), dict) else None), "bindings": [output_binding], }) status_rows.append({ "id": f"failures:item:{event_id}", "label": label, "value": {"status": event.get("status"), "error": event.get("error_message")}, "bindings": status_item_bindings, }) if not queries: query_bindings = [ _calculation_binding( resolver, binding_id="queries:empty", label="取数 Agent 查询事件计数", value=[], role="process", ) ] result_bindings = [ _calculation_binding( resolver, binding_id="raw:empty", label="取数 Agent 原始命中计数", value=[], role="output", ) ] status_bindings = [ _calculation_binding( resolver, binding_id="failures:empty", label="取数 Agent 失败查询计数", value=[], role="status", ) ] activity_rows: list[dict[str, Any]] = [] activity_bindings: list[SourceBinding] = [] for event in events: event_id = _integer(event.get("id")) or 0 binding = resolver.binding_for_event( event_id, binding_id=f"activities:event:{event_id}", role="process", ) activity_bindings.append(binding) activity_rows.append( { "id": f"activities:item:{event_id}", "label": str(event.get("event_name") or f"Event {event_id}"), "value": { "eventName": event.get("event_name"), "eventType": event.get("event_type"), "status": event.get("status"), }, "bindings": [binding], } ) return [ { "id": "task", "title": "完整取数任务", "presentation": "text", "business": {"value": task}, "bindings": [task_binding], "runtimeRefs": [f"event:{agent_id}"], "defaultOpen": True, }, { "id": "queries", "title": "查询过程", "presentation": "items", "business": {"items": query_rows}, "bindings": query_bindings, "runtimeRefs": _runtime_refs(query_bindings), "defaultOpen": True, }, { "id": "raw", "title": "原始命中", "presentation": "items", "business": {"items": result_rows}, "bindings": result_bindings, "runtimeRefs": _runtime_refs(result_bindings), "defaultOpen": True, }, { "id": "screening", "title": "初筛整理", "presentation": "text", "business": {"value": screening}, "bindings": [screening_binding], "runtimeRefs": [f"event:{agent_id}"], "defaultOpen": True, }, { "id": "failures", "title": "失败与空结果", "presentation": "items", "business": {"items": status_rows}, "bindings": status_bindings, "runtimeRefs": _runtime_refs(status_bindings), "defaultOpen": False, }, { "id": "activities", "title": "运行记录", "presentation": "process", "business": {"items": activity_rows}, "bindings": activity_bindings, "runtimeRefs": _runtime_refs(activity_bindings), "defaultOpen": False, }, ] def _calculation_binding( resolver: SourceResolver, *, binding_id: str, label: str, value: Any, role: str, ) -> SourceBinding: binding = resolver.binding_for_field( { "id": binding_id, "label": label, "value": value, "source": { "kind": "calculation", "label": label, "ref": f"calculation:{binding_id}", "fieldPath": "value", }, "relation": "persisted-output", "completeness": "complete", }, binding_id=binding_id, role=role, ) binding["evidence"]["confidence"] = "deterministic" return binding def _event_status_bindings( resolver: SourceResolver, event_id: int, *, prefix: str, include_output: bool = False, ) -> list[SourceBinding]: """Bind every Event value rendered by a status/diagnostic module.""" bindings = [ resolver.binding_for_event( event_id, binding_id=f"{prefix}:status:event:{event_id}", path="status", role="status", ), ] event = resolver.event_details.get(int(event_id)) or {} if "duration_ms" in event: bindings.append(resolver.binding_for_event( event_id, binding_id=f"{prefix}:duration:event:{event_id}", path="duration_ms", role="status", )) if pointer_value(event, "/error_message") is not None or "error_message" in event: bindings.append(resolver.binding_for_event( event_id, binding_id=f"{prefix}:error:event:{event_id}", path="error_message", role="status", )) if include_output: outcome = resolver.binding_for_event( event_id, binding_id=f"{prefix}:outcome:event:{event_id}", path="output.content", role="output", availability="returned-to-agent", ) outcome["transform"] = { "kind": "parsed", "operation": "retrieval-outcome-v1", "outputKey": "outcome", } outcome["evidence"]["confidence"] = "deterministic" bindings.append(outcome) completeness = resolver.binding_for_event( event_id, binding_id=f"{prefix}:completeness:event:{event_id}", path="", role="status", ) completeness["transform"] = { "kind": "parsed", "operation": "event-body-completeness-v1", "outputKey": "completeness", } completeness["evidence"]["confidence"] = "deterministic" bindings.append(completeness) return bindings def _business_modules(detail: dict[str, Any]) -> list[dict[str, Any]]: if isinstance(detail.get("blocks"), list): modules: list[dict[str, Any]] = [] for index, item in enumerate(detail.get("blocks") or [], 1): if not isinstance(item, dict): continue # Preserve the complete semantic block. Specialized business # blocks carry their real payload in routes/steps/branches/rows, # not in the generic value/items pair. module = dict(item) module["id"] = str(item.get("id") or f"block-{index}") module["title"] = item.get("title") or f"业务模块 {index}" if not module.get("items"): module["items"] = item.get("branches") or item.get("rows") modules.append(module) return modules return [ dict(item) for item in detail.get("businessSections") or [] if isinstance(item, dict) ] def _match_business_module( modules: list[dict[str, Any]], module_id: str, title: str ) -> dict[str, Any] | None: exact = next((item for item in modules if str(item.get("id")) == module_id), None) if exact: return exact normalized = _normalized_title(title) exact_title = next( (item for item in modules if _normalized_title(str(item.get("title") or "")) == normalized), None, ) if exact_title: return exact_title # Do not use substring similarity here. A nearby title is not proof that # two independently shaped modules contain the same business value. return None def _module_value(module: dict[str, Any] | None) -> Any: if not module: return None if module.get("content") is not None: return module.get("content") if module.get("value") is not None: return module.get("value") block_type = str(module.get("type") or "") if block_type == "implementation-plan": return { "mode": module.get("mode"), "routes": module.get("routes") or [], "reasoning": module.get("reasoning"), } if block_type == "creative-process": return module.get("steps") or [] if block_type == "evaluation-branches": return module.get("branches") or [] if module.get("rows") is not None: return module.get("rows") if module.get("routes") is not None: return module.get("routes") if module.get("steps") is not None: return module.get("steps") return None def _presentation(module: dict[str, Any] | None, title: str) -> str: raw = str((module or {}).get("presentation") or (module or {}).get("type") or "") if raw in {"items", "list"}: return "items" if any(token in title for token in ("规划", "路线", "计划")): return "plan" if any(token in title for token in ("过程", "处理", "调用")): return "process" if "评审" in title: return "review" if any(token in title for token in ("候选表", "主脚本", "产物")): return "artifact" if any(token in title for token in ("说明", "缺失", "状态", "完整性")): return "notice" return "text" def _runtime_refs(bindings: list[SourceBinding]) -> list[str]: refs: list[str] = [] for binding in bindings: source_id = str(binding.get("sourceId") or "") if source_id.startswith("event:"): refs.append(source_id) selector = binding.get("selector") or {} if selector.get("kind") == "members": refs.extend( str(member.get("sourceId") or "") for member in selector.get("members") or [] if str(member.get("sourceId") or "").startswith("event:") ) return list(dict.fromkeys(refs)) def _historical_implementation_task_modules( fields: dict[str, dict[str, Any]], resolver: SourceResolver, ) -> list[dict[str, Any]]: """Keep old Branch tasks readable without copying one blob into six modules.""" task_field = next( (field for field in fields.values() if str(field.get("label") or "") == "完整实现任务"), None, ) target_field = next( (field for field in fields.values() if str(field.get("label") or "") == "实现目标"), None, ) modules: list[dict[str, Any]] = [] if target_field and target_field.get("value") not in (None, ""): binding = resolver.binding_for_field(target_field, binding_id="scope:historical-target") modules.append({ "id": "scope", "title": "实现范围", "presentation": "text", "business": {"value": target_field.get("value")}, "bindings": [binding], "runtimeRefs": [], "defaultOpen": True, }) if task_field and task_field.get("value") not in (None, ""): binding = resolver.binding_for_field(task_field, binding_id="task:historical-full") modules.append({ "id": "historical-task", "title": "完整实现任务（历史回退）", "presentation": "text", "business": {"value": task_field.get("value")}, "bindings": [binding], "runtimeRefs": [], "defaultOpen": True, }) if modules: return modules binding = unresolved_binding( "historical-task:missing", reason="该历史方案没有保存实现 Agent 输入，也没有可用的 Branch 实现任务字段", code="source-record-not-saved", ) resolver.add_unresolved_source(binding) return [{ "id": "historical-task", "title": "实现任务记录缺失", "presentation": "notice", "business": {"value": "该历史方案未保存可恢复的完整实现任务。"}, "bindings": [binding], "runtimeRefs": [], "defaultOpen": True, }] def _multipath_comparison_binding( item: dict[str, Any], item_index: int, module_bindings: list[SourceBinding], resolver: SourceResolver, bundle: dict[str, Any], ) -> SourceBinding: """Bind one comparison row to its review, Branch status and final decision.""" text = str(item.get("value") or item.get("label") or "") match = re.search(r"方案\s*(\d+)", text) branch_id = int(match.group(1)) if match else None decision_id = next( ( _suffix_int(str(binding.get("sourceId") or "")) for binding in module_bindings if "script_build_multipath_decision" in str(binding.get("sourceId") or "") ), None, ) members: list[dict[str, Any]] = [] if branch_id is not None: branch = next( (row for row in bundle.get("branches") or [] if _integer(row.get("branch_id")) == branch_id), None, ) if branch is not None: members.append({ "id": f"comparison:{item_index}:branch", "label": f"方案 {branch_id} 处置状态", "value": branch.get("status"), "source": { "kind": "database", "label": "script_build_branch", "ref": f"round:{branch.get('round_index')}:branch:{branch_id}:output", "fieldPath": "status", }, "relation": "persisted-output", }) if decision_id is not None: decision = next( (row for row in bundle.get("multipathDecisions") or [] if _integer(row.get("id")) == decision_id), None, ) if decision is not None: members.append({ "id": f"comparison:{item_index}:decision", "label": "主 Agent 最终决定", "value": decision.get("decision"), "source": { "kind": "database", "label": "script_build_multipath_decision", "ref": f"multipath-decision:{decision_id}", "fieldPath": "decision", }, "relation": "persisted-output", }) for binding in module_bindings: source_id = str(binding.get("sourceId") or "") event_id = _suffix_int(source_id) if source_id.startswith("event:") else None if event_id is None: continue event = resolver.event_details.get(event_id) if event is None: continue members.append({ "id": f"comparison:{item_index}:review:{event_id}", "label": "多方案评审", "value": (event.get("output") or {}).get("content") if isinstance(event.get("output"), dict) else event.get("output"), "source": { "kind": "runtime-event", "label": str(event.get("event_name") or f"Event {event_id}"), "ref": f"event:{event_id}", "fieldPath": "output.content", }, "relation": "available-upstream", }) return resolver.members_binding( { "id": f"comparison:item:{item_index}", "label": str(item.get("label") or f"方案对照 {item_index}"), "value": item.get("value"), }, binding_id=f"comparison:item:{item_index}", members=members, reducer="multipath-review-and-decision-row-v1", role="basis", ) def _align_transforms_with_business( value: Any, bindings: list[SourceBinding], resolver: SourceResolver, module_id: str, ) -> None: """Make the transform describe what the first column actually renders.""" if value is None or not bindings: return if len(bindings) > 1: for binding in bindings: if binding.get("transform", {}).get("kind") == "direct": binding["transform"] = { "kind": "combined", "operation": f"business-module:{module_id}", } binding["evidence"]["confidence"] = "deterministic" return binding = bindings[0] if binding.get("transform", {}).get("kind") != "direct": return source = resolver.sources.get(str(binding.get("sourceId") or "")) or {} selector = binding.get("selector") or {} if selector.get("kind") == "json-pointer": selected = pointer_value(source.get("rawRecord"), str(selector.get("path") or "")) elif selector.get("kind") == "whole-record": selected = source.get("rawRecord") else: return if not _same_value(value, selected): binding["transform"] = { "kind": "parsed", "operation": "business-module-projection-v1", "outputKey": module_id, } binding["evidence"]["confidence"] = "deterministic" def _normalize_items(items: list[Any]) -> list[dict[str, Any]]: values: list[dict[str, Any]] = [] for index, item in enumerate(items, 1): if isinstance(item, dict): label = ( item.get("label") or item.get("title") or item.get("subject") or item.get("summary") or f"第 {index} 项" ) value = item.get("value") if "value" in item else item values.append({"label": label, "value": value}) else: values.append({"label": f"第 {index} 项", "value": item}) return values def _item_bindings( module_id: str, item_index: int, item: dict[str, Any], item_count: int, module_bindings: list[SourceBinding], resolver: SourceResolver, ) -> list[SourceBinding]: if not module_bindings: binding = unresolved_binding( f"{module_id}:item:{item_index}:unresolved", reason=f"子项「{item.get('label')}」没有可定位的来源绑定", code="binding-map-missing", ) resolver.add_unresolved_source(binding) return [binding] if len(module_bindings) == item_count and item_index <= len(module_bindings): binding = deepcopy(module_bindings[item_index - 1]) binding["id"] = f"{module_id}:item:{item_index}" source = resolver.sources.get(str(binding.get("sourceId") or "")) or {} selector = binding.get("selector") or {} selected = pointer_value(source.get("rawRecord"), str(selector.get("path") or "")) if not _same_value(item.get("value"), selected): binding["transform"] = { "kind": "parsed", "operation": "business-item-v1", "outputKey": str(item_index), } binding["evidence"]["confidence"] = "deterministic" return [binding] parent = module_bindings[0] selector = parent.get("selector") or {} source_id = str(parent.get("sourceId") or "") source_record = resolver.sources.get(source_id) or {} selected_parent = pointer_value( source_record.get("rawRecord"), str(selector.get("path") or "") ) if selector.get("kind") == "json-pointer" and isinstance(selected_parent, list): binding = deepcopy(parent) binding["id"] = f"{module_id}:item:{item_index}" binding["selector"] = { "kind": "json-pointer", "path": f"{str(selector.get('path') or '').rstrip('/')}/{item_index - 1}", } return [binding] event_id = _suffix_int(source_id) if source_id.startswith("event:") else None if event_id: role = str(parent.get("role") or "basis") path = str(selector.get("path") or "") if not path: path = "input.content.task" if role in {"input", "constraint", "basis"} else "output.content.summary" return [ resolver.text_span_binding( event_id, binding_id=f"{module_id}:item:{item_index}", path=path.lstrip("/").replace("/", "."), exact_text=_item_text(item.get("value")), role=role, ) ] binding = unresolved_binding( f"{module_id}:item:{item_index}:unresolved", reason=f"子项「{item.get('label')}」是结构化重组结果，无法安全定位到单一原始区间", source_id=source_id or None, resolution="unsafe", code="unsafe-association", ) if source_id and source_id in resolver.sources: return [binding] resolver.add_unresolved_source(binding) return [binding] def _item_text(value: Any) -> str: if isinstance(value, str): return value if not isinstance(value, dict): return str(value or "") for key in ("summary", "conclusion", "subject", "title", "evidence"): if value.get(key): return str(value[key]) return "" def _leaf_texts(value: Any) -> list[str]: """Return every visible textual leaf so a structured review item is fully bound.""" result: list[str] = [] if isinstance(value, str): cleaned = value.strip() if cleaned: result.append(cleaned) elif isinstance(value, dict): for child in value.values(): result.extend(_leaf_texts(child)) elif isinstance(value, list): for child in value: result.extend(_leaf_texts(child)) elif value is not None: result.append(str(value)) return list(dict.fromkeys(result)) def _same_value(left: Any, right: Any) -> bool: try: import json return json.dumps(left, ensure_ascii=False, sort_keys=True, default=str) == json.dumps( right, ensure_ascii=False, sort_keys=True, default=str ) except (TypeError, ValueError): return str(left) == str(right) def _unique_bindings(bindings: list[SourceBinding]) -> list[SourceBinding]: result: list[SourceBinding] = [] seen: set[str] = set() for binding in bindings: key = str(binding.get("id") or "") if key in seen: continue seen.add(key) result.append(binding) return result def _all_bindings(modules: list[dict[str, Any]]) -> list[SourceBinding]: result: list[SourceBinding] = [] for module in modules: result.extend(module.get("bindings") or []) business = module.get("business") or {} for item in business.get("items") or []: if isinstance(item, dict): result.extend(item.get("bindings") or []) return result def _used_source_ids(modules: list[dict[str, Any]]) -> set[str]: used: set[str] = set() def add_selector_members(selector: dict[str, Any]) -> None: if selector.get("kind") != "members": return for member in selector.get("members") or []: source_id = str(member.get("sourceId") or "") if source_id: used.add(source_id) child = member.get("selector") if isinstance(child, dict): add_selector_members(child) for module in modules: for binding in module.get("bindings") or []: source_id = str(binding.get("sourceId") or "") if source_id: used.add(source_id) selector = binding.get("selector") if isinstance(selector, dict): add_selector_members(selector) for source_id in module.get("runtimeRefs") or []: if isinstance(source_id, str) and source_id: used.add(source_id) return used def _aggregate_binding( card_kind: str, module_id: str, source_ids: list[str], fields: dict[str, dict[str, Any]], resolver: SourceResolver, bundle: dict[str, Any], ) -> SourceBinding | None: if not source_ids: return None target = next((fields[source_id] for source_id in source_ids if source_id in fields), None) if target is None: return None eligible = ( (card_kind == "round-result" and module_id in {"completion", "dispositions", "domain", "changes"}) or (card_kind == "round-summary" and module_id == "plan") or (card_kind == "final-result" and module_id in {"stats", "scale"}) or (card_kind == "domain-facts" and module_id in {"facts", "verification", "sources", "usage"}) or (card_kind == "retrieval-stage" and module_id in {"mode", "waves", "direct", "agents", "result"}) ) if not eligible: return None members: list[dict[str, Any]] = [] reducer = str((target.get("source") or {}).get("label") or "确定性聚合") if card_kind == "round-summary" and module_id == "plan": round_index = _round_index_from_field_id(str(target.get("id") or "")) row = next( ( item for item in bundle.get("rounds") or [] if _integer(item.get("round_index")) == round_index ), None, ) if row: for path in ("race_or_divide", "multipath_plan", "plan_note"): members.append( _member_field( "script_build_round", f"round:{round_index}", row, path, ) ) elif card_kind == "retrieval-stage": value = target.get("value") event_ids: list[int] = [] def collect_event_ids(node: Any) -> None: if isinstance(node, dict): event_id = _integer(node.get("eventId")) if event_id: event_ids.append(event_id) for child in node.values(): collect_event_ids(child) elif isinstance(node, list): for child in node: collect_event_ids(child) collect_event_ids(value) # Mode/wave/result are deterministic summaries of the whole scoped # retrieval stage, so their members are every Event loaded for this # card. Direct/Agent modules only retain the Events named by their # own structured projection. if module_id in {"mode", "waves", "result"}: event_ids.extend(resolver.event_details) for event_id in dict.fromkeys(event_ids): event = resolver.event_details.get(event_id) if not event: continue members.append( { "id": f"aggregate:retrieval:{module_id}:event:{event_id}", "label": str(event.get("event_name") or f"Event {event_id}"), "value": event, "source": { "kind": "runtime-event", "label": str(event.get("event_name") or f"Event {event_id}"), "ref": f"event:{event_id}", "fieldPath": "", }, "relation": "persisted-output", "completeness": "complete", } ) elif card_kind == "round-result": round_index = _round_index_from_field_id(str(target.get("id") or "")) if module_id in {"dispositions", "changes", "completion"}: for row in bundle.get("branches") or []: if _integer(row.get("round_index")) != round_index: continue branch_id = _integer(row.get("branch_id")) path = "status" if module_id == "dispositions" else "candidate_snapshot" members.append( { "id": f"aggregate:{module_id}:branch:{branch_id}", "label": f"方案 {branch_id}", "value": row.get(path), "source": { "kind": "database", "label": "script_build_branch", "ref": f"round:{round_index}:branch:{branch_id}:output", "fieldPath": path, }, "relation": "persisted-output", "completeness": "complete", } ) elif module_id == "domain": for row in bundle.get("domainInfo") or []: if _integer(row.get("round_index")) != round_index: continue members.append( { "id": f"aggregate:domain:{row.get('id')}", "label": "领域事实", "value": row, "source": { "kind": "database", "label": "script_build_domain_info", "ref": f"domain-info:{row.get('id')}", "fieldPath": "", }, "relation": "persisted-output", "completeness": "complete", } ) elif card_kind == "final-result" and module_id == "stats": for row in bundle.get("rounds") or []: members.append(_member_field("script_build_round", f"round:{row.get('round_index')}:result", row, "status")) for row in bundle.get("branches") or []: members.append(_member_field("script_build_branch", f"round:{row.get('round_index')}:branch:{row.get('branch_id')}:output", row, "status")) elif card_kind == "final-result" and module_id == "scale": artifact = bundle.get("currentArtifact") or {} for collection in ("paragraphs", "elements", "paragraphElements"): members.append( { "id": f"aggregate:artifact:{collection}", "label": collection, "value": artifact.get(collection) or [], "source": { "kind": "artifact", "label": "当前主脚本", "ref": "artifact:base:current", "fieldPath": collection, }, "relation": "persisted-output", "completeness": "complete", } ) elif card_kind == "domain-facts": match = re.search(r"branch:(\d+)", str(target.get("id") or "")) branch_id = _integer(match.group(1)) if match else None path = { "facts": "content", "verification": "note", "sources": "source", "usage": "content", }.get(module_id, "content") for row in bundle.get("domainInfo") or []: if _integer(row.get("branch_id")) != branch_id: continue members.append( { "id": f"aggregate:domain:{row.get('id')}:{path}", "label": f"领域事实 {row.get('id')}", "value": row.get(path), "source": { "kind": "database", "label": "script_build_domain_info", "ref": f"domain-info:{row.get('id')}", "fieldPath": path, }, "relation": "persisted-output", "completeness": "complete" if row.get(path) not in (None, "", [], {}) else "partial", } ) if not members and card_kind == "domain-facts": match = re.search(r"branch:(\d+)", str(target.get("id") or "")) branch_id = _integer(match.group(1)) if match else None return resolver.empty_database_result_binding( binding_id=f"{module_id}:empty-result", table="script_build_domain_info", filters={"branch_id": branch_id}, label="领域事实查询结果", ) if not members: return None return resolver.members_binding( target, binding_id=f"{module_id}:aggregate", members=members, reducer=reducer, ) def _member_field(table: str, ref: str, row: dict[str, Any], path: str) -> dict[str, Any]: return { "id": f"aggregate:{table}:{row.get('id')}:{path}", "label": f"{table}.{path}", "value": row.get(path), "source": { "kind": "database", "label": table, "ref": ref, "fieldPath": path, }, "relation": "persisted-output", "completeness": "complete", } def _round_index_from_field_id(value: str) -> int | None: match = re.search(r"round:(\d+)", value) return _integer(match.group(1)) if match else None def _business_notices(detail: dict[str, Any]) -> list[dict[str, Any]]: notices: list[dict[str, Any]] = [] for item in detail.get("notices") or []: if isinstance(item, dict) and item.get("message"): notices.append(item) if detail.get("missingActivityNotice"): notices.append({"level": "warning", "message": detail["missingActivityNotice"]}) return notices def _unique_notices(items: list[dict[str, Any]]) -> list[dict[str, Any]]: result: list[dict[str, Any]] = [] seen: set[str] = set() for item in items: message = str(item.get("message") or "") if not message or message in seen: continue seen.add(message) result.append({"level": item.get("level") or "info", "message": message}) return result def _fallback_card_data(detail_ref: str, business: dict[str, Any]) -> dict[str, Any]: modules = _business_modules(business) return { "cardKind": _fallback_card_kind(detail_ref), "completeness": "partial", "businessInputs": [], "businessOutputs": [], "runtime": {"summary": "", "units": []}, "displayUse": { "businessModules": [ {"id": item.get("id") or f"module-{index}", "title": item.get("title") or f"业务模块 {index}", "sourceIds": []} for index, item in enumerate(modules, 1) ], "cardFields": [], "unusedSourceIds": [], }, "notices": [ { "level": "warning", "message": "该历史卡片没有完整结构化来源投影，已保留业务详情并显式标记断链。", } ], } def _fallback_card_kind(detail_ref: str) -> str: if detail_ref.startswith("event:"): return "runtime-event" if detail_ref.startswith("round:"): return detail_ref.rsplit(":", 1)[-1] return detail_ref.split(":", 1)[0] or "unknown" def _event_text_path(event: dict[str, Any], side: str) -> str: wrapped = event.get(side) content = wrapped.get("content") if isinstance(wrapped, dict) else None if isinstance(content, dict): preferred = "task" if side == "input" else "summary" if isinstance(content.get(preferred), str): return f"{side}.content.{preferred}" return f"{side}.content" def _normalized_title(value: str) -> str: return re.sub(r"[\s/\u3001，,()（）]+", "", value).replace("完整", "") def _collect_event_ids(value: Any, ids: set[int], key: str = "") -> None: if isinstance(value, dict): for child_key, child in value.items(): lowered = str(child_key).lower() if lowered in { "eventid", "event_id", "implementereventid", "scopeeventid", "parenteventid", }: event_id = _integer(child) if event_id: ids.add(event_id) elif lowered in {"eventref", "currentrevisionref", "detailref"}: event_id = _suffix_int(str(child)) if event_id: ids.add(event_id) elif lowered in {"eventrefs", "technicalrefs", "revisionrefs"}: for ref in child or []: event_id = _suffix_int(str(ref)) if event_id: ids.add(event_id) _collect_event_ids(child, ids, lowered) elif isinstance(value, list): for child in value: _collect_event_ids(child, ids, key) def _view_branch(view: dict[str, Any], round_index: int, branch_id: int) -> dict[str, Any]: round_ = next( (item for item in view.get("rounds") or [] if _integer(item.get("roundIndex")) == round_index), {}, ) return next( (item for item in round_.get("branches") or [] if _integer(item.get("branchId")) == branch_id), {}, ) def _bundle_event(bundle: dict[str, Any], event_id: int | None) -> dict[str, Any] | None: return next( (item for item in bundle.get("events") or [] if _integer(item.get("id")) == event_id), None, ) def _integer(value: Any) -> int | None: try: return int(value) except (TypeError, ValueError): return None def _suffix_int(value: str) -> int | None: return _integer(str(value or "").rsplit(":", 1)[-1])