Reduce copies in agent invocation hot path

golem-common/src/schema/agent/mod.rs

@@ -40,7 +40,7 @@ use crate::schema::metadata::MetadataEnvelope;
 use crate::schema::schema_type::{NamedFieldType, SchemaType};
 use crate::schema::schema_value::SchemaValue;
 use crate::schema::validation::placement::validate_agent_type_placement;
-use crate::schema::validation::value::validate_value;
+use crate::schema::validation::value::{validate_record_fields, validate_value};
 use golem_schema_derive::{FromSchema, IntoSchema};
 use serde::{Deserialize, Serialize};
 use uuid::Uuid;
@@ -74,6 +74,12 @@ pub fn json_input_schema_value_to_typed_schema_value(
     graph: &SchemaGraph,
     input_schema: &InputSchema,
 ) -> Result<TypedSchemaValue, String> {
+    let result_graph = projected_input_graph(graph, input_schema);
+    validate_input_value(&result_graph, &value)?;
+    Ok(TypedSchemaValue::new(result_graph, value))
+}
+
+fn projected_input_graph(graph: &SchemaGraph, input_schema: &InputSchema) -> SchemaGraph {
     // Only user-supplied fields are part of the caller's input; auto-injected
     // fields (e.g. the principal) are filled by the host out of band and are
     // not present in the incoming value, so they are excluded from the record
@@ -89,18 +95,36 @@ pub fn json_input_schema_value_to_typed_schema_value(
         })
         .collect();
     let root = SchemaType::record(fields);
-    let result_graph = SchemaGraph {
+    SchemaGraph {
         defs: reachable_defs(graph, &root),
         root,
+    }
+}
+
+fn validate_input_value(graph: &SchemaGraph, value: &SchemaValue) -> Result<(), String> {
+    let validation = match (&graph.root, value) {
+        (
+            SchemaType::Record {
+                fields: field_types,
+                ..
+            },
+            SchemaValue::Record { fields: values },
+        ) => validate_record_fields(
+            graph,
+            field_types
+                .iter()
+                .map(|field| (field.name.as_str(), &field.body)),
+            values,
+        ),
+        _ => validate_value(graph, &graph.root, value),
     };
-    validate_value(&result_graph, &result_graph.root, &value).map_err(|errors| {
+    validation.map_err(|errors| {
         errors
             .into_iter()
             .map(|err| err.to_string())
             .collect::<Vec<_>>()
             .join("; ")
-    })?;
-    Ok(TypedSchemaValue::new(result_graph, value))
+    })
 }
 
 pub use crate::schema::graph::reachable_defs;
@@ -379,8 +403,8 @@ impl AgentMethodSchema {
     /// Validates the caller-supplied parameter record against this method's
     /// input schema and the owning agent's graph.
     pub fn validate_input(&self, graph: &SchemaGraph, input: &SchemaValue) -> Result<(), String> {
-        json_input_schema_value_to_typed_schema_value(input.clone(), graph, &self.input_schema)
-            .map(|_| ())
+        let input_graph = projected_input_graph(graph, &self.input_schema);
+        validate_input_value(&input_graph, input)
     }
 
     /// Returns whether a caller-supplied input or the output of this method can

golem-common/src/schema/agent/tests.rs

@@ -18,13 +18,14 @@ use crate::schema::agent::{
     AgentConfigDeclarationSchema, AgentConstructorSchema, AgentDependencySchema, AgentMethodSchema,
     AgentTypeSchema, AutoInjectedKind, FieldSource, InputSchema, NamedField, OutputSchema,
     ParsedAgentId, contains_stream_in_graph, json_input_schema_value_to_typed_schema_value,
-    typed_schema_value_with_projected_defs,
+    reachable_defs, typed_schema_value_with_projected_defs,
 };
 use crate::schema::graph::{SchemaGraph, SchemaTypeDef, TypedSchemaValue};
 use crate::schema::metadata::{MetadataEnvelope, TypeId};
 use crate::schema::schema_type::{NamedFieldType, SchemaType, SecretSpec, VariantCaseType};
 use crate::schema::schema_value::{SchemaValue, SecretValuePayload, VariantValuePayload};
 use crate::schema::stream::SchemaValueStream;
+use crate::schema::validation::validate_value;
 use proptest::prelude::*;
 use serde_json::json;
 use test_r::test;
@@ -775,15 +776,103 @@ fn method_input_validation_accepts_a_live_stream_handle() {
         )],
         OutputSchema::Unit,
     );
+    let stream = SchemaValueStream::from_host_endpoint(());
     let input = SchemaValue::Record {
-        fields: vec![SchemaValue::Stream(SchemaValueStream::from_host_endpoint(
-            (),
-        ))],
+        fields: vec![SchemaValue::Stream(stream.clone())],
     };
 
     method
         .validate_input(&SchemaGraph::empty(), &input)
         .unwrap();
+    assert!(stream.is_present());
+}
+
+#[test]
+fn method_input_validation_matches_owned_conversion() {
+    fn assert_matches_owned_conversion(
+        method: &AgentMethodSchema,
+        graph: &SchemaGraph,
+        input: SchemaValue,
+    ) {
+        let fields = method
+            .input_schema
+            .fields()
+            .iter()
+            .filter(|field| matches!(field.source, FieldSource::UserSupplied))
+            .map(|field| NamedFieldType {
+                name: field.name.clone(),
+                body: field.schema.clone(),
+                metadata: field.metadata.clone(),
+            })
+            .collect();
+        let root = SchemaType::record(fields);
+        let projected_graph = SchemaGraph {
+            defs: reachable_defs(graph, &root),
+            root,
+        };
+        let expected =
+            validate_value(&projected_graph, &projected_graph.root, &input).map_err(|errors| {
+                errors
+                    .into_iter()
+                    .map(|error| error.to_string())
+                    .collect::<Vec<_>>()
+                    .join("; ")
+            });
+        assert_eq!(method.validate_input(graph, &input), expected);
+    }
+
+    let scalar_method = method(
+        "compare",
+        vec![
+            NamedField::user_supplied("name", SchemaType::string()),
+            NamedField::auto_injected(
+                "principal",
+                AutoInjectedKind::Principal,
+                SchemaType::string(),
+            ),
+            NamedField::user_supplied("count", SchemaType::u64()),
+        ],
+        OutputSchema::Unit,
+    );
+    for input in [
+        SchemaValue::Record {
+            fields: vec![SchemaValue::String("ok".to_string()), SchemaValue::U64(1)],
+        },
+        SchemaValue::Record {
+            fields: vec![SchemaValue::String("missing-count".to_string())],
+        },
+        SchemaValue::Record {
+            fields: vec![
+                SchemaValue::U64(1),
+                SchemaValue::String("wrong".to_string()),
+            ],
+        },
+        SchemaValue::String("not-a-record".to_string()),
+    ] {
+        assert_matches_owned_conversion(&scalar_method, &SchemaGraph::empty(), input);
+    }
+
+    let graph = registry(vec![proj_def(
+        "Payload",
+        SchemaType::record(vec![proj_field("value", SchemaType::string())]),
+    )]);
+    let method = method(
+        "compare-ref",
+        vec![NamedField::user_supplied(
+            "payload",
+            SchemaType::ref_to(TypeId::new("Payload")),
+        )],
+        OutputSchema::Unit,
+    );
+    assert_matches_owned_conversion(
+        &method,
+        &graph,
+        SchemaValue::Record {
+            fields: vec![SchemaValue::Record {
+                fields: vec![SchemaValue::String("ok".to_string())],
+            }],
+        },
+    );
 }
 
 #[test]

golem-worker-executor/src/durable_host/mod.rs

@@ -4489,7 +4489,7 @@ impl<Ctx: WorkerCtx> InvocationHooks for DurableWorkerCtx<Ctx> {
                 _ => {}
             }
 
-            let (start_index, _) = self
+            let start_index = self
                 .public_state
                 .worker()
                 .oplog()