B1b: scope B1's anchor rule to the runtime domain; keep openhcs: keys resolvable

openhcs/core/callable_contract.py

@@ -1207,8 +1207,6 @@ def group_scope_inputs(self) -> ArtifactSpecCollection:
         input_specs = self.artifact_inputs
         source_refs = self.output_group_scope_sources
         if not source_refs:
-            if self.main_flow_supplies_primary_image:
-                return ArtifactSpecCollection(())
             return input_specs
         sources = ArtifactSpecCollection(
             source_spec
@@ -1228,12 +1226,23 @@ def group_scope_inputs(self) -> ArtifactSpecCollection:
             )
         return sources
 
+    @property
+    def lifecycle_anchor_owner_inputs(self) -> ArtifactSpecCollection:
+        """Inputs whose stored values, not main-flow anchors, drive invocation.
+
+        A callable fed main-flow pixels runs once per main-flow anchor; its
+        undeclared artifact inputs are context for that anchor. A declared
+        group-lineage relation still names its owning inputs.
+        """
+        if self.main_flow_supplies_primary_image and not self.output_group_scope_sources:
+            return ArtifactSpecCollection(())
+        return self.group_scope_inputs
+
     @property
     def main_flow_supplies_primary_image(self) -> bool:
         """Whether the runtime passes main-flow pixels as the primary argument.
 
-        Such a callable runs once per main-flow cohort; its artifact inputs are
-        context for that cohort, not the owner of its group scope.
+        Such a callable runs once per main-flow cohort.
         """
         adapter = self.runtime_adapter
         return self.accepts_implicit_main_flow_input and not (

openhcs/core/function_patterns.py

@@ -128,10 +128,10 @@ def from_invocation(
             for edge in invocation.artifact_input_edges
             if edge.storage_plan is not None
         )
-        group_scope_refs = invocation.contract.group_scope_inputs.ref_set()
+        owner_refs = invocation.contract.lifecycle_anchor_owner_inputs.ref_set()
         return (
             cls.ARTIFACT_MANAGED
-            if group_scope_refs and group_scope_refs <= stored_input_refs
+            if owner_refs and owner_refs <= stored_input_refs
             else cls.SOURCE_ANCHORED
         )
 

openhcs/processing/backends/lib_registry/openhcs_registry.py

@@ -767,8 +767,13 @@ def metadata_for_declared_callable(
             return None
         if _catalog_memory_types(func) is None:
             return None
+        registry = cls()
+        # An ``openhcs:`` key must resolve back through the catalog, which
+        # only scans this registry's own modules.
+        if not registry.owns_module(declared.__module__):
+            return None
 
-        return cls()._metadata_for_function(
+        return registry._metadata_for_function(
             func.__name__,
             func,
             _registry_catalog_module_for_callable(func),

openhcs/processing/backends/lib_registry/unified_registry.py

@@ -1507,6 +1507,15 @@ def _canonical_metadata_claims(
         if metadata is not None:
             yield metadata
 
+    def owns_module(self, module_name: str) -> bool:
+        """Whether ``module_name`` lies inside this registry's declared modules."""
+
+        return any(
+            module_name == module_pattern
+            or module_name.startswith(f"{module_pattern}.")
+            for module_pattern in self.get_module_patterns()
+        )
+
     def composite_keys_for_declared_callable(
         self,
         func: Callable,
@@ -1518,11 +1527,7 @@ def composite_keys_for_declared_callable(
             return (metadata.composite_key,)
 
         declared = inspect.unwrap(func)
-        if not any(
-            declared.__module__ == module_pattern
-            or declared.__module__.startswith(f"{module_pattern}.")
-            for module_pattern in self.get_module_patterns()
-        ):
+        if not self.owns_module(declared.__module__):
             return ()
 
         module_names = (

tests/unit/test_function_patterns.py

@@ -635,7 +635,7 @@ def assemble(image_tiles, positions):
     group = replace(compiled.default_group, invocations=(invocation,))
 
     assert invocation.contract.main_flow_supplies_primary_image
-    assert invocation.contract.group_scope_inputs == ArtifactSpecCollection(())
+    assert invocation.contract.lifecycle_anchor_owner_inputs == ArtifactSpecCollection(())
     assert group.runtime_domain is RuntimeInvocationDomain.SOURCE_ANCHORED
     anchors = ["z001", "z002", "z003"]
     assert group.runtime_domain.select_lifecycle_anchors(anchors) == anchors