Preserve custom-registration cause and original-owner observation

openhcs/agent/capabilities.py

@@ -78,6 +78,8 @@
 from openhcs.agent.dto.functions import (
     CustomFunctionRegistrationRequest,
     CustomFunctionRegistrationResult,
+    CustomFunctionRegistrationHandle,
+    CustomFunctionRegistrationObservation,
     FunctionCatalogPage,
     FunctionCatalogPreparationHandle,
     FunctionCatalogPreparationState,
@@ -2035,6 +2037,24 @@ class RegisterCustomFunctionCapability(FunctionCatalogCapability):
     )
 
 
+class ObserveCustomFunctionRegistrationCapability(FunctionCatalogCapability):
+    name = "openhcs_get_custom_function_registration_status"
+    cli_command = "get-custom-function-registration-status"
+    kind = CapabilityKind.TOOL
+    title = "Observe custom registration source"
+    description = "Read exact publication/persistence proofs through the original native owners using observation_handle. Does not evaluate/load source or prepare a catalog. Missing evidence stays not_observed; it never authorizes registration replay or proves original mutation did not occur."
+    service = "function_catalog"
+    exposition = FunctionCatalogCapability.exposition.refine(
+        visibility=CapabilityVisibility.STANDARD,
+    )
+    input_contract = CustomFunctionRegistrationHandle
+    output_contract = CustomFunctionRegistrationObservation
+    request_invocation = AgentDataclassRequestServiceInvocation(
+        service=lambda context: context.function_catalog,
+        method=lambda service, request: service.observe_custom_function_registration(request),
+    )
+
+
 class GetAuthoringContextCapability(KnowledgeCapability):
     name = "openhcs_get_authoring_context"
     cli_command = "authoring-context"

openhcs/agent/dto/functions.py

@@ -17,6 +17,7 @@
 
 from openhcs.agent.dto.common import (
     SCHEMA_VERSION,
+    AgentError,
     AgentDataclassCliRequest,
     AgentResultEnvelope,
 )
@@ -25,6 +26,7 @@
 from openhcs.agent.path_policy import AgentPathPolicy
 from openhcs.core.artifacts import ArtifactViewerStreaming
 from openhcs.core.function_reference import FunctionReference
+from openhcs.processing.custom_functions.source_namespace import CustomFunctionSource
 
 DEFAULT_FUNCTION_DETAIL_DOC_CHARS = 6_000
 
@@ -37,6 +39,7 @@ class FunctionCatalogControlMessageType(str, Enum):
     READ_DETAIL = "openhcs_function_detail_read"
     READ_REFERENCE = "openhcs_function_reference_read"
     REGISTER_CUSTOM = "openhcs_custom_function_register"
+    OBSERVE_CUSTOM_REGISTRATION = "openhcs_custom_function_registration_observe"
     CUSTOM_REGISTRATION_DESTINATION = "openhcs_custom_function_registration_destination"
     START_PREPARATION = "openhcs_function_catalog_prepare"
     READ_PREPARATION = "openhcs_function_catalog_preparation_status"
@@ -108,19 +111,83 @@ class FunctionSearchRequest(FunctionCatalogControlRequestABC):
 
 
 @dataclass(frozen=True, slots=True)
-class FunctionCatalogPreparationHandle(AgentDataclassCliRequest):
-    """The one preparation owner in this exact execution-server incarnation."""
+class FunctionCatalogOperationHandle(AgentDataclassCliRequest):
+    """Incarnation authority shared by catalog lifecycle operations."""
 
     connection: ExecutionConnectionSpec
     server_identity: ProcessIdentity
 
     def require_current_owner(self) -> None:
         if self.server_identity != ProcessIdentity.current():
             raise RuntimeError(
-                "Function catalog preparation owner changed; handle is stale."
+                "Function catalog operation owner changed; handle is stale."
             )
 
 
+@dataclass(frozen=True, slots=True)
+class FunctionCatalogPreparationHandle(FunctionCatalogOperationHandle):
+    """The one preparation owner in this exact execution-server incarnation."""
+
+
+@dataclass(frozen=True, slots=True)
+class CustomFunctionRegistrationHandle(FunctionCatalogOperationHandle):
+    """Content/owner proof for observing publication, not replay permission."""
+
+    content_sha256: str
+    function_name: str | None
+    persist: bool
+    storage_dir: str | None
+
+    @classmethod
+    def from_request(cls, request: CustomFunctionRegistrationRequest) -> Self:
+        return cls(
+            connection=request.connection, server_identity=request.require_server_identity(),
+            content_sha256=CustomFunctionSource.content_digest(request.source_code.encode("utf-8")),
+            function_name=request.function_name, persist=request.persist,
+            storage_dir=request.storage_dir,
+        )
+
+    def require_named_source(self) -> CustomFunctionSource:
+        if self.function_name is None:
+            raise ValueError("Persisted registration observation requires its admitted name.")
+        return CustomFunctionSource(self.function_name, self.content_sha256)
+
+    def require_storage_dir(self) -> Path:
+        """Decode this handle's admitted source store once for native/path owners."""
+        if self.storage_dir is None:
+            raise ValueError("Persisted registration observation requires its admitted store.")
+        return Path(self.storage_dir)
+
+
+class CustomFunctionRegistrationObservationOutcome(str, Enum):
+    """Current canonical proofs; absence never establishes a failed mutation."""
+
+    NOT_OBSERVED = "not_observed"
+    PUBLISHED = "published"
+    PERSISTED_ONLY = "persisted_only"
+    REGISTERED = "registered"
+
+    @classmethod
+    def from_proofs(cls, *, published: bool, persisted: bool, persist: bool) -> Self:
+        if published:
+            return cls.REGISTERED if persisted or not persist else cls.PUBLISHED
+        return cls.PERSISTED_ONLY if persisted else cls.NOT_OBSERVED
+
+
+@dataclass(frozen=True, kw_only=True)
+class CustomFunctionRegistrationObservation(AgentResultEnvelope):
+    handle: CustomFunctionRegistrationHandle
+    published_sources: tuple[CustomFunctionSource, ...] = ()
+    persisted_source: CustomFunctionSource | None = None
+    outcome: CustomFunctionRegistrationObservationOutcome = field(init=False)
+
+    def __post_init__(self) -> None:
+        object.__setattr__(self, "outcome", CustomFunctionRegistrationObservationOutcome.from_proofs(
+            published=bool(self.published_sources), persisted=self.persisted_source is not None,
+            persist=self.handle.persist,
+        ))
+
+
 class FunctionCatalogPreparationOutcome(str, Enum):
     NOT_STARTED = "not_started"
     PENDING = "pending"
@@ -263,6 +330,12 @@ class CustomFunctionRegistrationDestinationRequest(FunctionCatalogControlRequest
     message_type = FunctionCatalogControlMessageType.CUSTOM_REGISTRATION_DESTINATION
 
 
+@dataclass(frozen=True, slots=True)
+class CustomFunctionRegistrationObservationRequest(FunctionCatalogControlRequestABC):
+    handle: CustomFunctionRegistrationHandle
+    message_type = FunctionCatalogControlMessageType.OBSERVE_CUSTOM_REGISTRATION
+
+
 @dataclass(frozen=True, slots=True)
 class CustomFunctionRegistrationDestination:
     storage_dir: str
@@ -660,6 +733,25 @@ class CustomFunctionRegistrationResult(AgentResultEnvelope):
     next_steps: tuple[str, ...] = ()
     connection: ExecutionConnectionSpec = field(default_factory=ExecutionConnectionSpec)
     server_identity: ProcessIdentity | None = None
+    observation_handle: CustomFunctionRegistrationHandle | None = None
+
+    @classmethod
+    def uncertain(
+        cls, request: CustomFunctionRegistrationRequest, error: Exception,
+    ) -> Self:
+        """Preserve the actual cause and same-owner read-only recovery identity."""
+        handle = CustomFunctionRegistrationHandle.from_request(request)
+        cause = error.to_agent_error() if isinstance(error, AgentFacingErrorMixin) else AgentError.from_exception(
+            "custom_function_registration_uncertain", error,
+            hint="Preserve the original request. Observe this handle; do not replay registration. Absence is not proof of no mutation.",
+        )
+        return cls(
+            schema_version=SCHEMA_VERSION, persisted=request.persist,
+            storage_dir=request.storage_dir, connection=request.connection,
+            server_identity=request.server_identity, observation_handle=handle,
+            errors=(cause,),
+            next_steps=("Observe this observation_handle with openhcs_get_custom_function_registration_status; do not replay source.",),
+        )
 
 
 class CustomFunctionRegistrationControlResponse(
@@ -686,6 +778,13 @@ def destination(self) -> CustomFunctionRegistrationDestination:
         return self.value
 
 
+class CustomFunctionRegistrationObservationControlResponse(
+    FunctionCatalogControlResponseBase[CustomFunctionRegistrationObservation]
+):
+    field = FunctionCatalogControlField.RESULT
+    value_type = CustomFunctionRegistrationObservation
+
+
 def catalog_page(
     *,
     items: tuple[FunctionCatalogEntry, ...],

openhcs/agent/services/endpoint_function_catalog_service.py

@@ -19,6 +19,8 @@
     CustomFunctionRegistrationDestinationRequest,
     CustomFunctionRegistrationRequest,
     CustomFunctionRegistrationResult,
+    CustomFunctionRegistrationHandle,
+    CustomFunctionRegistrationObservation,
     FunctionCatalogControlRequest,
     FunctionCatalogEntry,
     FunctionCatalogPage,
@@ -32,7 +34,6 @@
     FunctionReferenceControlRequest,
     FunctionSearchRequest,
 )
-from openhcs.agent.exceptions import AgentFacingErrorMixin
 from openhcs.agent.path_policy import AgentPathPolicy
 from openhcs.agent.services.function_catalog_service import FunctionCatalogServiceABC
 from openhcs.runtime.zmq_config import OpenHCSZMQConfig
@@ -47,19 +48,6 @@
 logger = logging.getLogger(__name__)
 
 
-class CustomFunctionRegistrationUncertainError(AgentFacingErrorMixin, RuntimeError):
-    agent_error_code = "custom_function_registration_uncertain"
-    agent_error_hint = "Preserve this request and endpoint receipt. Do not replay or fall back; persistence may have completed."
-
-    def __init__(self, request: CustomFunctionRegistrationRequest):
-        super().__init__(
-            f"Registration observation failed after invoking {request.connection.transport_endpoint()}; "
-            f"destination={request.storage_dir}, function={request.function_name!r}. "
-            f"Selected server={request.server_identity!r}. "
-            "Outcome is uncertain; source or registry mutation may have completed."
-        )
-
-
 FunctionCatalogClientFactory = Callable[
     [OpenHCSZMQConfig],
     "FunctionCatalogExecutionClient",
@@ -471,8 +459,18 @@ def register_custom_function(
             result = client.register_custom_function(
                 request, operation_deadline=deadline
             )
+            if result.errors:
+                if result.observation_handle != CustomFunctionRegistrationHandle.from_request(request):
+                    raise ValueError("Registration error returned a different observation owner.")
+                return result
             destination.require_result(request, result)
-            if not request.persist:
+        except FunctionCatalogEndpointUnavailableError:
+            # Native admission failed before the source-bearing exchange.
+            raise
+        except Exception as error:
+            return CustomFunctionRegistrationResult.uncertain(request, error)
+        if not request.persist:
+            try:
                 from openhcs.processing.custom_functions.manager import (
                     CustomFunctionManager,
                 )
@@ -483,13 +481,29 @@ def register_custom_function(
                     clear_caches=False,
                     emit_signal=False,
                 )
-            self.invalidate()
-        except FunctionCatalogEndpointUnavailableError:
-            # Native admission failed before the source-bearing exchange, so this
-            # is recoverable rejection, not an uncertain mutation receipt.
-            raise
-        except Exception as error:
-            raise CustomFunctionRegistrationUncertainError(request) from error
+            except Exception as error:
+                from openhcs.agent.dto.common import AgentError
+                # Native receipt is already confirmed; do not erase it on local failure.
+                return replace(result, errors=(AgentError.from_exception(
+                    "custom_function_local_projection_failed", error,
+                    hint="Native registration is confirmed. Preserve its receipt; do not register again to repair local projection.",
+                ),))
+        self.invalidate()
+        return result
+
+    def observe_custom_function_registration(
+        self, handle: CustomFunctionRegistrationHandle,
+    ) -> CustomFunctionRegistrationObservation:
+        if handle.persist:
+            from openhcs.processing.custom_functions.manager import CustomFunctionManager
+            source = handle.require_named_source()
+            self._path_policy.assert_readable_location(CustomFunctionManager.source_path_for_name(
+                handle.require_storage_dir(), source.function_name,
+            ))
+        endpoint = self._endpoint_for_connection(handle.connection)
+        result = self._client_for(endpoint).observe_custom_function_registration(handle)
+        if result.handle != handle:
+            raise ValueError("Registration observation returned a different owner/source handle.")
         return result
 
     def _endpoint_for_connection(

openhcs/agent/services/function_catalog_service.py

@@ -37,6 +37,8 @@
     CustomFunctionRegistrationDestinationRequest,
     CustomFunctionRegistrationRequest,
     CustomFunctionRegistrationResult,
+    CustomFunctionRegistrationHandle,
+    CustomFunctionRegistrationObservation,
     FunctionArtifactSpec,
     FunctionCatalogEntry,
     FunctionCatalogPage,
@@ -599,6 +601,12 @@ def register_custom_function(
     ) -> CustomFunctionRegistrationResult:
         """Register one custom declaration through this catalog authority."""
 
+    @abstractmethod
+    def observe_custom_function_registration(
+        self, handle: CustomFunctionRegistrationHandle,
+    ) -> CustomFunctionRegistrationObservation:
+        """Observe exact existing publication/persistence without evaluating source."""
+
     @abstractmethod
     def search(
         self,
@@ -755,6 +763,37 @@ def admit_write(path: Path) -> Path:
             ),
             connection=request.connection,
             server_identity=server_identity,
+            observation_handle=CustomFunctionRegistrationHandle.from_request(request),
+        )
+
+    def observe_custom_function_registration(
+        self, handle: CustomFunctionRegistrationHandle,
+    ) -> CustomFunctionRegistrationObservation:
+        from openhcs.agent.dto.common import AgentWarning
+        from openhcs.processing.custom_functions.runtime_registry import CustomFunctionRuntimeRegistry
+
+        handle.require_current_owner()
+        published = CustomFunctionRuntimeRegistry.published_sources_for_content(
+            handle.content_sha256, function_name=handle.function_name,
+        )
+        persisted = None
+        warnings = ()
+        if handle.persist:
+            manager = custom_function_manager.CustomFunctionManager(create_storage=False)
+            if handle.require_storage_dir().resolve(strict=False) != manager.storage_dir.resolve(strict=False):
+                raise ValueError("Registration observation reached a different native source store.")
+            source = handle.require_named_source()
+            self._path_policy.assert_readable_location(manager.source_path_for_name(manager.storage_dir, source.function_name))
+            try:
+                manager.require_source(source)
+            except (OSError, RuntimeError) as error:
+                warnings = (AgentWarning("registration_source_not_observed", str(error),
+                    "Missing/changed current source is not proof of no original mutation; do not replay."),)
+            else:
+                persisted = source
+        return CustomFunctionRegistrationObservation(
+            schema_version=SCHEMA_VERSION, handle=handle, published_sources=published,
+            persisted_source=persisted, warnings=warnings,
         )
 
     def custom_function_registration_destination(