WIP: consume explicit ObjectState field states

openhcs/agent/services/config_service.py

@@ -23,7 +23,8 @@
 )
 
 from metaclass_registry import AutoRegisterMeta
-from objectstate import get_base_type_for_lazy
+from objectstate.declarations import InheritedDefault
+from objectstate.ui_visibility import UIParameterVisibilityPolicy
 from formbridge.services.parameter_help_service import (
     class_docstring_text,
     dataclass_parameter_descriptions,
@@ -64,6 +65,7 @@
 from openhcs.core.steps.abstract import AbstractStep
 from openhcs.core.steps.function_step import FunctionStep
 from openhcs.core.axes import AxisFamily, GroupingDeclaration
+from python_introspect import resolve_declared_type
 
 AgentConfig: TypeAlias = GlobalPipelineConfig | PipelineConfig
 
@@ -811,8 +813,8 @@ def _schema_for_field(
     source_type = lazy_base or field_type
     nested_types = _nested_dataclass_types(field_type)
     ui_hidden = False
-    if "ui_hidden" in field.metadata:
-        ui_hidden = bool(field.metadata["ui_hidden"])
+    if UIParameterVisibilityPolicy in field.metadata:
+        ui_hidden = UIParameterVisibilityPolicy.field_hidden(field)
     default_repr, default_origin = _default_schema(field)
     return ConfigFieldSchema(
         path=field.name if field_path is None else field_path,
@@ -832,8 +834,7 @@ def _schema_for_field(
         lazy=lazy_base is not None,
         inheritable=(
             inheritable
-            or "_inherited_default" in field.metadata
-            or "_inherited_default_factory" in field.metadata
+            or InheritedDefault in field.metadata
         ),
         declaring_type=_type_repr(_field_declaring_type(declaring_cls, field.name)),
         default_origin=default_origin,
@@ -948,7 +949,8 @@ def _selected_config_type_schemas(
 def _lazy_base_type(field_type) -> type | None:
     if not isinstance(field_type, type):
         return None
-    return get_base_type_for_lazy(field_type)
+    base_type = resolve_declared_type(field_type)
+    return base_type if base_type is not field_type else None
 
 
 def _type_repr(field_type) -> str:
@@ -958,12 +960,9 @@ def _type_repr(field_type) -> str:
 
 
 def _default_schema(field) -> tuple[str | None, str | None]:
-    inherited_default = field.metadata.get("_inherited_default", MISSING)
-    inherited_factory = field.metadata.get("_inherited_default_factory", MISSING)
-    if inherited_default is not MISSING:
-        return _default_value_repr(inherited_default), "inherited_default"
-    if inherited_factory is not MISSING:
-        return _factory_default_repr(inherited_factory), "inherited_factory"
+    if InheritedDefault in field.metadata:
+        inherited = field.metadata[InheritedDefault]
+        return _default_value_repr(inherited.value()), inherited.schema_origin
     if field.default is not MISSING:
         return _default_value_repr(field.default), "field_default"
     if field.default_factory is not MISSING:
@@ -998,7 +997,6 @@ def _parameter_default_repr(parameter: inspect.Parameter) -> str | None:
 
 def _field_declaring_type(cls: type, field_name: str) -> type:
     source_cls = _lazy_base_type(cls) or cls
-    inherited_metadata = frozenset(("_inherited_default", "_inherited_default_factory"))
     fallback = source_cls
     for candidate in source_cls.__mro__:
         if field_name not in candidate.__dict__.get("__annotations__", {}):
@@ -1012,7 +1010,7 @@ def _field_declaring_type(cls: type, field_name: str) -> type:
         )
         if candidate_field is None:
             continue
-        if not inherited_metadata.intersection(candidate_field.metadata):
+        if InheritedDefault not in candidate_field.metadata:
             return candidate
     return fallback
 

openhcs/agent/services/object_state_field_help_service.py

@@ -2,10 +2,8 @@
 
 from __future__ import annotations
 
-import importlib
 from collections.abc import Callable, Mapping, Sequence
 from dataclasses import dataclass
-from types import ModuleType
 
 from formbridge.services.parameter_help_service import (
     NO_PARAMETER_DESCRIPTION,
@@ -138,7 +136,7 @@ def describe(
             )
 
         try:
-            help_target = self._import_qualified_target(field.object_state_path_type)
+            help_target = field.declared_by.resolve()
             parameter_info = UnifiedParameterAnalyzer.analyze(help_target).get(
                 field.field_name
             )
@@ -158,7 +156,7 @@ def describe(
             parameter_description = self._description_with_function_parameter_context(
                 parameter_content.description,
                 help_target,
-                field.object_state_path_type,
+                field.declared_by.import_path,
                 field.field_name,
                 request.max_description_chars,
             )
@@ -176,8 +174,6 @@ def describe(
                 schema_version=SCHEMA_VERSION,
                 address=field.address,
                 field=field,
-                object_type=scope.object_type,
-                help_target_type=field.object_state_path_type,
                 parameter_name=field.field_name,
                 target_summary=target_docstring.summary,
                 target_description=target_docstring.description,
@@ -289,45 +285,18 @@ def _scope_and_field(
             return scope, None
         return None, None
 
-    @staticmethod
-    def _import_qualified_target(
-        qualified_name: str,
-    ) -> type | Callable[..., object]:
-        module_name, _, qualname = qualified_name.partition(":")
-        if not qualname:
-            module_name, _, qualname = qualified_name.rpartition(".")
-        if not module_name or not qualname:
-            raise ValueError(f"Help target name is not importable: {qualified_name!r}")
-        module = importlib.import_module(module_name)
-        symbol = ObjectStateFieldHelpService._resolve_qualname(module, qualname)
-        if not isinstance(symbol, type) and not callable(symbol):
-            raise TypeError(
-                f"Imported symbol is not a class or callable: {qualified_name!r}"
-            )
-        return symbol
-
-    @staticmethod
-    def _resolve_qualname(module: ModuleType, qualname: str) -> object:
-        symbol: object = module
-        for name in qualname.split("."):
-            symbol = vars(symbol)[name]
-        return symbol
-
     def _description_with_value_context(
         self,
         parameter_description: str | None,
         field: UiObjectStateFieldSummary,
         max_description_chars: int,
     ) -> str | None:
-        callable_context = self._callable_value_context(
-            field.resolved_value,
-            max_description_chars=max_description_chars,
+        describe = lambda value: self._callable_value_context(
+            value, max_description_chars=max_description_chars
         )
+        callable_context = field.resolved_value.project(describe)
         if callable_context is None:
-            callable_context = self._callable_value_context(
-                field.raw_value,
-                max_description_chars=max_description_chars,
-            )
+            callable_context = field.raw_value.project(describe)
         if callable_context is None:
             return parameter_description
         if parameter_description:

openhcs/agent/services/pipeline_authoring_service.py

@@ -8,7 +8,6 @@
 from itertools import count
 from typing import TypeAlias
 
-from objectstate import get_base_type_for_lazy
 
 from openhcs.agent.dto.common import SCHEMA_VERSION, RenderedSource
 from formbridge.contracts import AgentError, AgentWarning
@@ -45,6 +44,7 @@
 )
 from openhcs.core.steps.abstract import AbstractStep
 from openhcs.core.steps.function_step import FunctionStep
+from python_introspect import resolve_declared_type
 
 StepConfig: TypeAlias = object
 PIPELINE_EMPTY_WARNING = AgentWarning(
@@ -475,6 +475,6 @@ def _step_config_class_from_patch(patch: ConfigPatch) -> type[StepConfig]:
 
 def _instantiate_config_patch(patch: ConfigPatch) -> StepConfig:
     config_class = _step_config_class_from_patch(patch)
-    coercion_class = get_base_type_for_lazy(config_class) or config_class
+    coercion_class = resolve_declared_type(config_class)
     values = coerce_dataclass_patch_values(coercion_class, patch.values)
     return config_class(**values)

openhcs/core/axis_filter.py

@@ -5,6 +5,7 @@
 from dataclasses import dataclass
 from collections.abc import Mapping
 from typing import TYPE_CHECKING, TypeAlias
+from python_introspect import resolve_declared_type
 
 if TYPE_CHECKING:
     from openhcs.core.config import PartitionFilterConfig, PartitionFilterMode
@@ -53,13 +54,12 @@ def resolution_for(
         self,
         config: "PartitionFilterConfig",
     ) -> StepAxisFilterResolution | None:
-        from objectstate import get_base_type_for_lazy
 
-        config_type = get_base_type_for_lazy(type(config)) or type(config)
+        config_type = resolve_declared_type(type(config))
         config_mro = config_type.mro()
         matches: list[tuple[int, StepAxisFilterResolution]] = []
         for owner_type, resolution in self.resolutions_by_config_type.items():
-            owner_base = get_base_type_for_lazy(owner_type) or owner_type
+            owner_base = resolve_declared_type(owner_type)
             if not isinstance(owner_base, type):
                 continue
             if owner_base not in config_mro: