{"record":{"id":"7410adfd877adf29","repo":"deepset-ai/haystack","slug":"type-conflict-for-input-socket-name-from-compo","errorCode":null,"errorMessage":"Type conflict for input '{socket_name}' from component '{comp_name}'. Existing type: {existing_socket_info['type']}, new type: {socket_info['type']}.","messagePattern":"Type conflict for input '(.+?)' from component '(.+?)'\\. Existing type: (.+?), new type: (.+?)\\.","errorType":"exception","errorClass":"InvalidMappingTypeError","httpStatus":null,"severity":"error","filePath":"haystack/core/super_component/super_component.py","lineNumber":247,"sourceCode":"        \"\"\"\n        aggregated_inputs: dict[str, dict[str, Any]] = {}\n        for wrapper_input_name, pipeline_input_paths in input_mapping.items():\n            for path in pipeline_input_paths:\n                comp_name, socket_name = self._split_component_path(path)\n                socket_info = pipeline_inputs[comp_name][socket_name]\n\n                # Add to aggregated inputs\n                existing_socket_info = aggregated_inputs.get(wrapper_input_name)\n                if existing_socket_info is None:\n                    aggregated_inputs[wrapper_input_name] = {\"type\": socket_info[\"type\"]}\n                    if not socket_info[\"is_mandatory\"]:\n                        aggregated_inputs[wrapper_input_name][\"default\"] = _delegate_default\n                    continue\n\n                is_compatible, common_type = _is_compatible(existing_socket_info[\"type\"], socket_info[\"type\"])\n\n                if not is_compatible:\n                    raise InvalidMappingTypeError(\n                        f\"Type conflict for input '{socket_name}' from component '{comp_name}'. \"\n                        f\"Existing type: {existing_socket_info['type']}, new type: {socket_info['type']}.\"\n                    )\n\n                # Use the common type for the aggregated input\n                aggregated_inputs[wrapper_input_name][\"type\"] = common_type\n\n                # If any socket requires mandatory inputs then the aggregated input is also considered mandatory.\n                # So we use the type of the mandatory input and remove the default value if it exists.\n                if socket_info[\"is_mandatory\"]:\n                    aggregated_inputs[wrapper_input_name].pop(\"default\", None)\n\n        return aggregated_inputs\n\n    @staticmethod\n    def _create_input_mapping(pipeline_inputs: dict[str, dict[str, Any]]) -> dict[str, list[str]]:\n        \"\"\"\n        Create an input mapping from pipeline inputs.","sourceCodeStart":229,"sourceCodeEnd":265,"githubUrl":"https://github.com/deepset-ai/haystack/blob/e318778c9bf60a1963e3b5f451359655dd696c30/haystack/core/super_component/super_component.py#L229-L265","documentation":"When multiple pipeline paths map to the same wrapper input, their socket types must be compatible; otherwise the SuperComponent cannot determine a single input type and raises InvalidMappingTypeError with both conflicting types.","triggerScenarios":"input_mapping = {\"q\": [\"retriever.query\", \"ranker.query\"]} where one expects str and the other expects List[Document], and no common type exists.","commonSituations":"Aggregating inputs from components with mismatched typed sockets; pipeline edits changed a socket's type annotation; connecting str and int-typed sockets.","solutions":["Align the socket types: change the components' socket types or wrap with a component that converts types","Map the conflicting paths to different wrapper input names instead of aggregating them","Only group paths whose sockets share a compatible type"],"exampleFix":"// before\ninput_mapping = {\"q\": [\"retriever.query\", \"ranker.documents\"]}  # str vs List[Document]\n// after\ninput_mapping = {\"query\": [\"retriever.query\"], \"documents\": [\"ranker.documents\"]}","handlingStrategy":"validation","validationCode":"from haystack.core.super_component.super_component import _is_compatible\n\ndef check_no_type_conflicts(pipe, input_mapping):\n    inputs = pipe.inputs()\n    seen = {}\n    for name, paths in input_mapping.items():\n        for p in paths:\n            comp, sock = p.split(\".\")\n            t = inputs[comp][sock][\"type\"]\n            if name in seen:\n                ok, _ = _is_compatible(seen[name], t)\n                if not ok:\n                    raise TypeError(f\"Conflicting types for '{name}': {seen[name]} vs {t}\")\n            seen[name] = t","typeGuard":"def types_are_compatible(t1, t2) -> bool:\n    from haystack.core.super_component.super_component import _is_compatible\n    ok, _ = _is_compatible(t1, t2)\n    return ok","tryCatchPattern":"from haystack.core.errors import InvalidMappingTypeError\ntry:\n    super_comp = SuperComponent(pipeline=pipe, input_mapping=input_mapping)\nexcept InvalidMappingTypeError as e:\n    if \"Type conflict for input\" in str(e):\n        print(\"Split into separate wrapper inputs or align socket types:\", e)\n    else:\n        raise","preventionTips":["Only aggregate paths whose sockets share the same or compatible types","Split conflicting paths into distinct wrapper input names","Keep component socket type annotations consistent across the pipeline"],"tags":["super-component","type-conflict","input-mapping"],"backgroundTag":"socket-type-conflict","analyzedSha":"e318778c9bf60a1963e3b5f451359655dd696c30","analyzedAt":"2026-08-30T11:45:20.711Z","schemaVersion":2},"datasetVersion":"2026-08-30T13:17:10.514Z"}