Declare nothing for a scope that composes nothing, and keep what exists

Porting Modelplane to reference.composing turned up two gaps.

A scope that read a reference but didn't compose its resource, because
the value it needed wasn't there yet, still declared the dependency.
Crossplane ignores a dependency whose resource it doesn't know about, and
says so in an event on every reconcile. A scope now declares
dependencies only for a resource that's composed or already exists.

And a resource that exists, but that the block didn't compose because
the value was missing, was left out of desired state, which deletes it.
The fallback only covered a resource the block had written. It now keeps
an existing resource either way: its current spec under whatever the
block wrote, or its spec alone if the block wrote nothing, with a warning
saying why. That's the natural way to write it -

    secrets = c.ref(cluster.status.secrets)
    if secrets:
        compose_backend(secrets)

- and it shouldn't delete the backend when the cluster's secrets
briefly disappear.

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

Author: stevendborrelli

crossplane/function/reference.py

@@ -671,35 +671,51 @@ def _get(self, s: _Source, path: list, what: str) -> typing.Any:
         return v
 
     def _close(self) -> None:
-        _record(self.rsp, self.name, self._sources)
-
-        if not self._unresolved:
-            return
-
         exists = self.name in self.req.observed.resources
-        ordered = request.has_capability(self.req, fnv1.CAPABILITY_DEPENDENCIES)
 
-        if not exists:
-            # Crossplane won't create the resource until the dependency is
-            # ready. One that doesn't enforce dependencies would create it
-            # without the field, so hold it back by leaving it out instead.
-            if not ordered:
+        if self._unresolved:
+            if exists:
+                self._keep_current_spec()
+            elif not request.has_capability(self.req, fnv1.CAPABILITY_DEPENDENCIES):
+                # Crossplane won't create the resource until the dependency
+                # is ready. One that doesn't enforce dependencies would create
+                # it without the field, so hold it back by leaving it out.
                 self.rsp.desired.resources.pop(self.name, None)
-            return
 
-        # The resource exists, but something it refers to doesn't. The fields
-        # that refer to it came back None and were left out, and applying
-        # that would unset them. Keep what the resource already has for any
-        # field this function no longer sets, and say so.
+        # Declare dependencies only for a resource that's composed or exists.
+        # A scope that composed nothing has nothing to order, and Crossplane
+        # would ignore the dependencies, with an event saying so.
+        if self.name in self.rsp.desired.resources or exists:
+            _record(self.rsp, self.name, self._sources)
+
+    def _keep_current_spec(self) -> None:
+        """Keep an existing resource's spec while a reference it needs is gone.
+
+        The fields that refer to what's missing came back None and were left
+        out, and applying that would unset them. A function that didn't
+        compose the resource at all, because the value it needed wasn't
+        there, would have it deleted. Either way, keep what the resource
+        already has for anything this function doesn't set, and say so.
+        """
+        observed = resource.struct_to_dict(
+            self.req.observed.resources[self.name].resource
+        )
         if self.name in self.rsp.desired.resources:
             r = self.rsp.desired.resources[self.name]
-            observed = resource.struct_to_dict(
-                self.req.observed.resources[self.name].resource
-            )
             body = resource.struct_to_dict(r.resource)
             if "spec" in observed:
                 body["spec"] = _overlay(observed["spec"], body.get("spec", {}))
             r.resource.CopyFrom(resource.dict_to_struct(body))
+        else:
+            meta = observed.get("metadata", {})
+            body = {
+                "apiVersion": observed.get("apiVersion"),
+                "kind": observed.get("kind"),
+                "metadata": {k: meta[k] for k in ("name", "namespace") if k in meta},
+            }
+            if "spec" in observed:
+                body["spec"] = observed["spec"]
+            resource.update(self.rsp.desired.resources[self.name], body)
 
         response.warning(
             self.rsp,
@@ -761,7 +777,12 @@ def composing(
     left out. If the resource doesn't exist yet that's what ordering is for:
     Crossplane waits for the dependency before creating it. If it does exist,
     it keeps its current spec for the fields that were left out, and the
-    response carries a warning saying why.
+    response carries a warning saying why. That holds even if the block
+    doesn't compose the resource at all because the value it needed is
+    missing: an existing resource is kept rather than deleted.
+
+    Dependencies are declared only for a resource that's composed or already
+    exists. A block that composes nothing declares nothing.
     """
     scope = Scope(req, rsp, name)
     yield scope

tests/test_reference.py

@@ -581,6 +581,52 @@ def test_existing_resource_keeps_its_spec(self) -> None:
         self.assertEqual(rsp.results[0].severity, fnv1.SEVERITY_WARNING)
         self.assertIn("vpc.status.atProvider.id", rsp.results[0].message)
 
+    def test_composing_nothing_declares_nothing(self) -> None:
+        req = fnv1.RunFunctionRequest(meta=ORDERED)
+        rsp = response.to(req)
+        vpc = reference.named("vpc", VPC)
+
+        with reference.composing(req, rsp, "subnet") as c:
+            vpc_id = c.ref(vpc.status.atProvider.id)
+            if vpc_id:
+                desired_subnet(rsp, vpc_id)
+
+        self.assertNotIn("subnet", rsp.desired.resources)
+        self.assertEqual(edges(rsp), [])
+
+    def test_existing_resource_not_composed_is_kept(self) -> None:
+        subnet = {
+            "apiVersion": "ec2.aws.m.upbound.io/v1beta1",
+            "kind": "Subnet",
+            "metadata": {"name": "xr-subnet-1", "uid": "abc", "resourceVersion": "7"},
+            "spec": {"forProvider": {"region": "us-east-1", "vpcId": "vpc-0123"}},
+            "status": {"atProvider": {"id": "subnet-9"}},
+        }
+        req = fnv1.RunFunctionRequest(meta=ORDERED, observed=observed(subnet=subnet))
+        rsp = response.to(req)
+        vpc = reference.named("vpc", VPC)
+
+        with reference.composing(req, rsp, "subnet") as c:
+            vpc_id = c.ref(vpc.status.atProvider.id)
+            if vpc_id:
+                desired_subnet(rsp, vpc_id)
+
+        # Kept rather than deleted, as its spec, without what the API server
+        # owns: uid, resourceVersion and status.
+        self.assertEqual(
+            body(rsp, "subnet"),
+            {
+                "apiVersion": "ec2.aws.m.upbound.io/v1beta1",
+                "kind": "Subnet",
+                "metadata": {"name": "xr-subnet-1"},
+                "spec": {"forProvider": {"region": "us-east-1", "vpcId": "vpc-0123"}},
+            },
+        )
+        self.assertEqual(
+            edges(rsp), [{"resource": "subnet", "composed_resource": "vpc"}]
+        )
+        self.assertEqual(rsp.results[0].severity, fnv1.SEVERITY_WARNING)
+
     def test_without_capability_holds_back_by_omission(self) -> None:
         req = fnv1.RunFunctionRequest(meta=UNORDERED)
         rsp = response.to(req)
@@ -628,6 +674,9 @@ def test_required(self) -> None:
 
         with reference.composing(req, rsp, "app-config") as c:
             host = c.ref(db.status.address)
+            c.update(
+                {"apiVersion": "v1", "kind": "ConfigMap", "data": {"DB_HOST": host}}
+            )
 
         self.assertEqual(host, "db.a")
         self.assertEqual(