Return a typed reference's value as its model types it

A resource named with its model gives c.ref the field's type, and the
signature promises it: c.ref(cluster.status.secrets) is typed
list[Secret]. It returned the JSON observed state stores instead, a list
of dicts, so code written against the type failed at runtime, and code
written against the JSON failed type checking. Porting Modelplane hit
both.

c.ref now validates the value against the field's type with a pydantic
TypeAdapter, so a nested object comes back as its model and a list of
them as a list of models. A value that doesn't fit is an error naming
the field, rather than a model built from the wrong data. Without a
model there's no type to fit, and the value is the JSON as before.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

Author: stevendborrelli

crossplane/function/reference.py

@@ -609,13 +609,26 @@ def ref(self, value: V) -> V:
             state, and the dependency means Crossplane doesn't create this
             resource until the value exists.
 
+            A resource named with its model returns the value as the model
+            types it: a nested object comes back as its model, and a list of
+            them as a list of models, rather than as the JSON observed state
+            stores. Without a model the value is the JSON.
+
         Raises:
             TypeError: value isn't a field of a named resource.
-            ValueError: value is the whole resource rather than a field of it.
+            ValueError: value is the whole resource rather than a field of
+                it, or the observed value doesn't fit the model's type for it.
         """
         marker = _decode(ref(value))
         assert marker is not None  # noqa: S101  # ref always returns a marker.
-        return typing.cast(V, self._get(*marker, value._describe()))
+        v = self._get(*marker, value._describe())
+        if v is None or value._model is None:
+            return typing.cast(V, v)
+        try:
+            return typing.cast(V, pydantic.TypeAdapter(value._model).validate_python(v))
+        except pydantic.ValidationError as e:
+            msg = f"{value._describe()} doesn't match its model: {e}"
+            raise ValueError(msg) from e
 
     def external_name(self, named_resource: typing.Any) -> str | None:
         """Read a named resource's external name, and depend on that resource.

tests/test_reference.py

@@ -88,6 +88,8 @@ def observed(**resources: dict) -> fnv1.State:
     "status": {"atProvider": {"id": "vpc-0123", "arn": "arn:aws:ec2:vpc/vpc-0123"}},
 }
 
+POOL_OBSERVED = {"ranges": [{"from": "10.0.0.0"}, {"from": "10.1.0.0"}]}
+
 ORDERED = fnv1.RequestMeta(
     capabilities=[fnv1.CAPABILITY_CAPABILITIES, fnv1.CAPABILITY_DEPENDENCIES]
 )
@@ -514,6 +516,49 @@ def test_returns_values_and_records_edges(self) -> None:
         )
         self.assertEqual(rsp.results, [])
 
+    def test_typed_values_come_back_as_models(self) -> None:
+        req = fnv1.RunFunctionRequest(
+            meta=ORDERED, observed=observed(pool=POOL_OBSERVED)
+        )
+        rsp = response.to(req)
+        pool = reference.named("pool", Pool)
+
+        with reference.composing(req, rsp, "subnet") as c:
+            ranges = c.ref(pool.ranges)
+            first = c.ref(pool.ranges[0])
+            start = c.ref(pool.ranges[0].from_)
+
+        self.assertEqual(
+            ranges, [Range(**{"from": "10.0.0.0"}), Range(**{"from": "10.1.0.0"})]
+        )
+        self.assertIsInstance(first, Range)
+        self.assertEqual(start, "10.0.0.0")
+
+    def test_untyped_values_come_back_as_json(self) -> None:
+        req = fnv1.RunFunctionRequest(
+            meta=ORDERED, observed=observed(pool=POOL_OBSERVED)
+        )
+        rsp = response.to(req)
+        pool = reference.named("pool")
+
+        with reference.composing(req, rsp, "subnet") as c:
+            ranges = c.ref(pool.ranges)
+
+        self.assertEqual(ranges, [{"from": "10.0.0.0"}, {"from": "10.1.0.0"}])
+
+    def test_values_that_dont_fit_the_model_fail(self) -> None:
+        req = fnv1.RunFunctionRequest(
+            meta=ORDERED, observed=observed(pool={"ranges": [{"from": 7}]})
+        )
+        rsp = response.to(req)
+        pool = reference.named("pool", Pool)
+
+        with (
+            self.assertRaisesRegex(ValueError, "pool.ranges doesn't match its model"),
+            reference.composing(req, rsp, "subnet") as c,
+        ):
+            c.ref(pool.ranges)
+
     def test_works_in_fields_that_are_not_strings(self) -> None:
         vpc_body = {"spec": {"forProvider": {"enableDnsSupport": True}}}
         req = fnv1.RunFunctionRequest(meta=ORDERED, observed=observed(vpc=vpc_body))