def compose_manifests(
left: GraphManifest,
right: GraphManifest,
op: ComposeManifestsOp,
*,
bump_version: bool | Literal["minor"] = "minor",
finish_init: bool = True,
strict_references: bool = False,
dynamic_edge_feedback: bool = False,
) -> GraphManifest:
"""Return a new manifest that is the deterministic compose of *left* and *right*.
Applies explicit equivalences in *op* (property alignment → boundary rename →
union of schema / resources / bindings → merge equivalent types). Does not
invent semantic matches.
"""
if not isinstance(op, ComposeManifestsOp):
raise TypeError(
f"compose_manifests expects ComposeManifestsOp, got {type(op)!r}"
)
out_left = left.model_copy(deep=True)
out_right = right.model_copy(deep=True)
left_schema = _require_schema(out_left, "left")
right_schema = _require_schema(out_right, "right")
left_vertices = left_schema.core_schema.vertex_config.vertex_set
right_vertices = right_schema.core_schema.vertex_config.vertex_set
for veq in op.vertices:
if veq.left not in left_vertices:
raise ValueError(
f"compose_manifests: left vertex {veq.left!r} not in left manifest"
)
if veq.right not in right_vertices:
raise ValueError(
f"compose_manifests: right vertex {veq.right!r} not in right manifest"
)
_align_side(out_left, side="left", op=op)
_align_side(out_right, side="right", op=op)
left_resource_names: set[str] = set()
if out_left.ingestion_model is not None:
left_resource_names = {r.name for r in out_left.ingestion_model.resources}
_apply_right_resource_policy(out_right, op, left_resource_names)
_apply_right_schema_collision_policy(out_left, out_right, op)
composed_schema = _union_schema(
_require_schema(out_left, "left"),
_require_schema(out_right, "right"),
op,
)
composed_ingestion = _union_ingestion(
out_left.ingestion_model, out_right.ingestion_model
)
composed_bindings = _union_bindings(
out_left.bindings, out_right.bindings, name_conflict=op.name_conflict
)
result = GraphManifest(
graph_schema=composed_schema,
ingestion_model=composed_ingestion,
bindings=composed_bindings,
)
_bump_schema_version(result, bump_version)
if finish_init:
result.finish_init(
strict_references=strict_references,
dynamic_edge_feedback=dynamic_edge_feedback,
)
return result