[Brook] Added generic call methods to the Worker extension.

crates/sdk-core/CHANGELOG.md

@@ -41,6 +41,9 @@ relevant information.
   seconds on `after_complete`. The replay worker never calls the hooks.
 * The info a `WorkflowTaskInterceptor` gets has the run's last completed Workflow Task and the
   tasks of the run that history shows as failed or timed out.
+* A `WorkerExtension` has `call` and `call_async`, which take a method name and encoded input. A
+  bridge can route a generic lang call to them, so the extension needs no bridge function of its
+  own. Core never calls them.
 
 ### Security
 * Ephemeral server downloads now use `zip` 9.0 for ZIP extraction, replacing the 8.x dependency

crates/sdk-core/src/lib.rs

@@ -51,7 +51,7 @@ pub use worker::{
     WorkflowErrorType, WorkflowSlotKind,
 };
 pub use worker::{
-    RunCloseKind, WorkerExtension, WorkflowTaskCompletionInfo, WorkflowTaskEpoch,
+    ExtensionError, RunCloseKind, WorkerExtension, WorkflowTaskCompletionInfo, WorkflowTaskEpoch,
     WorkflowTaskFailedInfo, WorkflowTaskInterceptError, WorkflowTaskInterceptor,
 };
 

crates/sdk-core/src/worker/extension.rs

@@ -8,7 +8,10 @@ use std::{
 
 /// A component that lang passes to the worker through [crate::WorkerConfig::extension].
 ///
-/// Core never inspects the extension beyond the hooks it asks for here.
+/// Core never inspects the extension beyond the hooks it asks for here. Core never calls
+/// [WorkerExtension::call] or [WorkerExtension::call_async]. A bridge exposes them to lang, so lang
+/// code reaches the extension in the same native library without a bridge function of its own.
+#[async_trait::async_trait]
 pub trait WorkerExtension: Send + Sync + 'static {
     /// A name that identifies the extension in logs.
     fn name(&self) -> &str;
@@ -18,6 +21,56 @@ pub trait WorkerExtension: Send + Sync + 'static {
     fn workflow_task_interceptor(&self) -> Option<Arc<dyn WorkflowTaskInterceptor>> {
         None
     }
+
+    /// Runs `method` with encoded input and returns the encoded output. It must not block, since
+    /// lang may call it from workflow code.
+    fn call(&self, method: &str, _input: &[u8]) -> Result<Vec<u8>, ExtensionError> {
+        Err(ExtensionError::unknown_method(method))
+    }
+
+    /// Like [WorkerExtension::call], for work that waits on I/O.
+    async fn call_async(&self, method: String, _input: Vec<u8>) -> Result<Vec<u8>, ExtensionError> {
+        Err(ExtensionError::unknown_method(&method))
+    }
+}
+
+/// The error of a [WorkerExtension] call. It has the fields of a gRPC status, so a bridge can turn
+/// it into the same lang error as a failed service call.
+#[derive(Clone, Debug, thiserror::Error)]
+#[error("{code:?}: {message}")]
+#[non_exhaustive]
+pub struct ExtensionError {
+    /// The status code.
+    pub code: tonic::Code,
+    /// The message for the caller.
+    pub message: String,
+    /// Encoded error details, empty if there are none.
+    pub details: Vec<u8>,
+}
+
+impl ExtensionError {
+    /// Creates an error with no details.
+    pub fn new(code: tonic::Code, message: impl Into<String>) -> Self {
+        Self {
+            code,
+            message: message.into(),
+            details: vec![],
+        }
+    }
+
+    /// Adds encoded error details.
+    pub fn with_details(mut self, details: Vec<u8>) -> Self {
+        self.details = details;
+        self
+    }
+
+    /// The error for a method the extension does not have.
+    pub fn unknown_method(method: &str) -> Self {
+        Self::new(
+            tonic::Code::Unimplemented,
+            format!("Unknown extension method {method}"),
+        )
+    }
 }
 
 /// Hooks around each `RespondWorkflowTaskCompleted` call that Core makes for a run.
@@ -150,3 +203,72 @@ impl WorkflowTaskInterceptError {
         }
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    struct Named;
+
+    impl WorkerExtension for Named {
+        fn name(&self) -> &str {
+            "named"
+        }
+    }
+
+    #[tokio::test]
+    async fn calls_to_an_extension_without_methods_are_unimplemented() {
+        let extension: Arc<dyn WorkerExtension> = Arc::new(Named);
+        let err = extension.call("Append", b"x").unwrap_err();
+        assert_eq!(err.code, tonic::Code::Unimplemented);
+        assert!(err.message.contains("Append"));
+        let err = extension
+            .call_async("Append".to_string(), vec![])
+            .await
+            .unwrap_err();
+        assert_eq!(err.code, tonic::Code::Unimplemented);
+    }
+
+    struct Echo;
+
+    #[async_trait::async_trait]
+    impl WorkerExtension for Echo {
+        fn name(&self) -> &str {
+            "echo"
+        }
+
+        fn call(&self, method: &str, input: &[u8]) -> Result<Vec<u8>, ExtensionError> {
+            match method {
+                "Echo" => Ok(input.to_vec()),
+                _ => {
+                    Err(ExtensionError::new(tonic::Code::NotFound, "no stream")
+                        .with_details(vec![7]))
+                }
+            }
+        }
+
+        async fn call_async(
+            &self,
+            method: String,
+            input: Vec<u8>,
+        ) -> Result<Vec<u8>, ExtensionError> {
+            self.call(&method, &input)
+        }
+    }
+
+    #[tokio::test]
+    async fn calls_reach_the_extension_methods() {
+        let extension: Arc<dyn WorkerExtension> = Arc::new(Echo);
+        assert_eq!(extension.call("Echo", b"abc").unwrap(), b"abc");
+        assert_eq!(
+            extension
+                .call_async("Echo".to_string(), b"abc".to_vec())
+                .await
+                .unwrap(),
+            b"abc"
+        );
+        let err = extension.call("Read", b"").unwrap_err();
+        assert_eq!(err.code, tonic::Code::NotFound);
+        assert_eq!(err.details, [7]);
+    }
+}

crates/sdk-core/src/worker/mod.rs

@@ -12,7 +12,7 @@ mod workflow;
 pub(crate) const PAYLOADS_TOO_LARGE_FAILURE_TYPE: &str = "PayloadsTooLarge";
 
 pub use extension::{
-    RunCloseKind, WorkerExtension, WorkflowTaskCompletionInfo, WorkflowTaskEpoch,
+    ExtensionError, RunCloseKind, WorkerExtension, WorkflowTaskCompletionInfo, WorkflowTaskEpoch,
     WorkflowTaskFailedInfo, WorkflowTaskInterceptError, WorkflowTaskInterceptor,
 };
 use temporalio_client::{Connection, PayloadErrorLimits, worker::NamespaceDescriptionSource};