fix: de-duplicate identical pipelines when combining Processing objects

[seanmcculloch]

Currently, the metadata manager merges Processing from multiple upstream sources, that usually share a pipeline, and it used to de-dup those itself. Moving that aggregation onto the schema's "+" operator and from_metadata lost the de-dup, since Processing.add concatenates pipelines without removing duplicates. This restores it in add (keyed on full identifiers.Code identity/__eq__, not name), so every + consumer gets it and the manager can drop its local shim.

What changed

Processing.__add__ concatenated pipelines without de-duplication, so combining N Processing objects that share a pipeline produced N identical Code entries. It now removes duplicates after merging.

remove_duplicates gains an equality-based fallback so it handles unhashable elements such as pydantic Code models, not just hashable scalars.

Review notes

• Dedup preserves first-seen order and runs only when the merged list is non-empty.
• Distinct pipelines that share a name are kept — identity is full model equality, not name.

[seanmcculloch]

Eg, removing duplicate aind_data_schema.components.identifiers.Code objects like pipelines

[dbirman]

How can this happen -- also isn't this a risk if someone runs the same code in multiple branches of a pipeline?

[seanmcculloch]

@dbirman We could end up with duplicate entries in the Processing.pipelines list easily during the step of aggregating processing metadata from the different capsules in the pipeline.

de-duping used to be handled my the metadata manager: https://github.com/AllenNeuralDynamics/aind-metadata-manager/blob/3d342db6a8aea4a5e157d0686de0ed913dd7a3ce/src/aind_metadata_manager/metadata_manager.py#L416-L440

I am currently working on this PR, which drops a lot of aggregation logic in the metadata manager, in favor or relying on the data schema itself (through __add__ and from_metadata), and we should de-duplicate this list.

BUT: it is risky in a way - since the description notes that Processing.pipelines is only matched via DataProcess.pipeline_name

aind-data-schema/src/aind_data_schema/core/processing.py

Lines 105 to 112 in f38cb0f

	pipelines: Optional[List[Code]] = Field(
	default=None,
	title="Pipelines",
	description=(
	"For processing done with pipelines, list the repositories here. Pipelines must use the name field "
	",and be referenced in the pipeline_name field of a DataProcess."
	),
	)

However, Code.commit_hash and Code.version are optional, so I'm not sure we can make the linking stronger without enforcing that they exist.. So it is indeed risky for different commits of the same pipeline.

[dbirman]

Individual capsules in a pipeline shouldn't be writing the pipeline though? How would they "know" that they are inside of a pipeline? It's the manager during aggregation that should assign them all the pipeline_name and add that object.

[dbirman]

I'll go a bit farther than my comment above -- I don't think this is logic that the schema should be implementing. I think this kind of management has to happen downstream of the merge code, which we should try to keep as minimal as possible.

[seanmcculloch]

@dbirman The intention is that this will be called by the metadata-manager, as it is right now. just that this PR is moving that code from the metadata-manager into the data-schema itself. Perhaps it should stay in the metadata-manager for now then?

The change here itself only removes exact duplicates within the Processing.pipelines: List[Code].

Can you confirm that this PR should be closed and that logic should stay in the metadata-manager?

[dbirman]

I think the situation that this PR is solving for, which is roughly: "Two identical pipelines were run in parallel and are now being merged" is a situation that should never happen, which is why I'm pushing back.

I think that's actually not what you're trying to fix though, you're trying to fix: "A single pipeline has capsules whose processing needs to be merged, but each of the capsules wrote their own pipeline information". That's a bug in the capsules -- capsules (in my opinion) should not write pipeline information at all. It's only at the last step in the pipeline when all capsules are done that the metadata manager should capture all the individual capsule processing and then wrap them all with the pipeline info.

Does that make more sense? Hop on a call if it's not clear.