feat: let a preloaded deployment skip its universe load, and deploy_env load the universe

src/agent_env/env/env.py

@@ -74,6 +74,7 @@ class DeployedEnv:
     # From the card whenever the record carries it (__post_init__); stored because readers of the JSON use them.
     mcp_url: Optional[str] = None
     mcp_server_name: Optional[str] = None
+    preloaded_universe: Optional[dict] = None
     # Set by the registry and the deploy step, never by a provider.
     metadata: Optional[dict] = None
     instance_id: Optional[str] = None
@@ -134,6 +135,7 @@ def _fields_from(cls, data: dict) -> dict:
             instance_id=data.get("instance_id"),
             created_at_utc=data.get("created_at_utc"),
             expires_at_utc=data.get("expires_at_utc"),
+            preloaded_universe=data.get("preloaded_universe"),
         )
 
 
@@ -188,6 +190,7 @@ def _fields_from(cls, data: dict) -> dict:
 @dataclass
 class LoadEnvironmentUniverseArtifactResult:
     metadata_filepaths: dict[str, str] = field(default_factory=dict)
+    delivered_preloaded: bool = False
     # Which path the load actually took. A snapshot restore swaps a pre-ingested servicedb
     # image; a miss re-ingests every service over HTTP under a 600s per-service cap. The
     # two are comparable in wall-clock on a well-sized VM (measured ~4min vs ~3.5min on a

src/agent_env/env/envs/multi_env.py

@@ -157,6 +157,7 @@ async def load_environment_artifact(self, environment_artifact: EnvironmentArtif
         # A built-in's deploy, or a caller wiring one up by hand, leaves a gateway; a plugin's leaves only its record.
         if self._gateway_url is None and self._deployed is None:
             raise RuntimeError("Environment not deployed - call deploy() first")
+        self._void_preloaded_universe()
         # MCP servers are checked first; if a website shares the same environment_name, the MCP server wins.
         for env in self.mcp_server_envs:
             if env.environment_name == environment_artifact.environment_name:
@@ -172,6 +173,7 @@ async def load_environment_universe_artifact(
         self,
         environment_universe_artifact: EnvironmentUniverseArtifact,
         snapshot_after_load: bool | None = None,
+        skip_if_preloaded: bool = False,
     ) -> LoadEnvironmentUniverseArtifactResult:
         """Seed this env from a universe artifact, preferring a pre-baked snapshot.
 
@@ -192,6 +194,10 @@ async def load_environment_universe_artifact(
         and pushes a multi-GB servicedb image (~5min, ~2.3GB stored) — worth it to make a
         universe cheap to re-serve, not worth it on a one-off deploy. The bake is
         best-effort and never fails a load that already succeeded.
+
+        ``skip_if_preloaded`` skips the data load for a deployment its provider delivered
+        with this universe already loaded and untouched since (see ``_delivered_with``). It
+        is off by default, so a caller's load always loads unless the caller asks otherwise.
         """
         import uuid
         from agent_env.env.env import LoadEnvironmentUniverseArtifactResult
@@ -212,17 +218,26 @@ async def load_environment_universe_artifact(
         # Keyed on the service images rather than self.id, so re-registering the same
         # servers under a new env id still hits (and a rebuilt server correctly misses).
         env_fingerprint = compute_env_fingerprint(self) if can_restore else None
+        preloaded = skip_if_preloaded and await self._delivered_with(environment_universe_artifact)
+        if not preloaded:
+            self._void_preloaded_universe()
         snapshot = get_env_snapshot_store().get_clean(
             self.id,
             environment_universe_artifact.id,
             environment_universe_artifact.version,
             env_fingerprint=env_fingerprint,
-        ) if can_restore else None
+        ) if can_restore and not preloaded else None
         restored_from_snapshot = False
         snapshot_baked: bool | None = None
         snapshot_bake_error: str | None = None
 
-        if snapshot:
+        if preloaded:
+            logger.info(
+                "env=%s was delivered with universe %s v%s already loaded and untouched; "
+                "skipping the data load",
+                self.id, environment_universe_artifact.id, environment_universe_artifact.version,
+            )
+        elif snapshot:
             logger.info(f"Found clean snapshot for env={self.id} universe={environment_universe_artifact.id}: artifact={snapshot.db_image_artifact_id} v{snapshot.db_image_artifact_version} instance={snapshot.instance_id}")
             await self._load_from_snapshot(snapshot)
             restored_from_snapshot = True
@@ -334,7 +349,7 @@ async def _load_one(idx: int, artifact):
         # image swap. Deliberately placed here: after the universe stamp, which
         # EnvSnapshot.create reads back off the instance record, and before the metadata
         # download, so the captured PGDATA reflects the data load and nothing else.
-        if not restored_from_snapshot and can_restore:
+        if not restored_from_snapshot and not preloaded and can_restore:
             if snapshot_after_load is None:
                 snapshot_after_load = _snapshot_after_load_default()
             if snapshot_after_load:
@@ -361,12 +376,60 @@ async def _load_one(idx: int, artifact):
                 logger.info(f"Successfully downloaded metadata '{key}'")
         return LoadEnvironmentUniverseArtifactResult(
             metadata_filepaths=metadata_filepaths,
+            delivered_preloaded=preloaded,
             restored_from_snapshot=restored_from_snapshot,
             snapshot_db_image_artifact_id=snapshot.db_image_artifact_id if snapshot else None,
             snapshot_baked=snapshot_baked,
             snapshot_bake_error=snapshot_bake_error,
         )
 
+    async def _delivered_with(self, environment_universe_artifact) -> bool:
+        """Whether this deployment was handed over with this universe already loaded, and
+        nothing has touched its data since.
+
+        Only the provider that made the deployment can say so, by setting
+        ``preloaded_universe`` on the record it returns: a warm pool, say, that loaded the
+        universe before anyone asked for it. This is what lets a complete load be skipped
+        where resume never skips one. Resume can't tell a caller's explicit load from a
+        retry, but here the deployment itself guarantees the data is the universe asked
+        for. The changelog gate still applies, so data touched since breaks the guarantee
+        and the universe loads as usual.
+        """
+        preloaded = getattr(self._deployed, "preloaded_universe", None) or {}
+        if (preloaded.get("id"), preloaded.get("version")) != (
+            environment_universe_artifact.id, environment_universe_artifact.version,
+        ):
+            return False
+        if self._sandbox is None or self._sandbox.mode == SANDBOX_MODE_CONTAINER:
+            return False
+        try:
+            from agent_env.env.snapshot_store import _check_changelog_empty
+
+            return await _check_changelog_empty(self._sandbox)
+        except Exception as e:  # noqa: BLE001
+            logger.warning("Could not confirm the preloaded universe is untouched (%s); loading it", e)
+            return False
+
+    def _void_preloaded_universe(self) -> None:
+        """Drop the deployment's preloaded guarantee before a load replaces its data.
+
+        A load reinstalls each service's changelog triggers, which empties the changelog.
+        Left standing, a guarantee for universe A would survive loading B and pass the
+        changelog gate on the next request for A, skipping a load the data needs. Cleared on
+        the stored record too, so a reattach can't revive it.
+        """
+        if not getattr(self._deployed, "preloaded_universe", None):
+            return
+        self._deployed.preloaded_universe = None
+        if not self._instance_id:
+            return
+        try:
+            from agent_env.env.store import get_env_instance_store
+
+            get_env_instance_store().clear_preloaded_universe(self._instance_id)
+        except Exception as e:  # noqa: BLE001
+            logger.warning("Could not void the preloaded universe on instance %s: %s", self._instance_id, e)
+
     async def _resumable_environments(self, environment_universe_artifact, total: int) -> set[str]:
         """Services safe to skip because a previous attempt already loaded them.
 

src/agent_env/env/store.py

@@ -300,6 +300,14 @@ def clear_loaded_environments(self, instance_id: str) -> None:
             UpdateSpec(set={"loaded_environments": {}}),
         )
 
+    def clear_preloaded_universe(self, instance_id: str) -> None:
+        """Void the universe an instance was delivered with, once a load replaces its data."""
+        self._doc_store.update(
+            ENV_INSTANCES_COLLECTION,
+            Filter.of(instance_id=instance_id),
+            UpdateSpec(set={"preloaded_universe": None}),
+        )
+
     def merge_metadata(self, instance_id: str, values: dict) -> None:
         """Merge key/values into an instance record's ``metadata``.
 

src/agent_env/providers/env_providers/env_provider.py

@@ -33,6 +33,7 @@ class EnvironmentProvider(ABC):
     """Deploys an env and hands back its record; tears down what it created."""
 
     type: ClassVar[str]  # the name an env declares it by, and that its records carry as env_provider_type
+    delivers_universe: ClassVar[bool] = False
 
     @abstractmethod
     async def deploy(self, env: Env, sandbox_provider: SandboxProvider) -> DeployedEnv:

src/agent_env/task_step/task_steps/deploy_env.py

@@ -120,6 +120,29 @@ def preflight(self) -> list[str]:
             return [f"deploy_env '{self.id}': env '{self.env_id}' can't be loaded: {e}"]
         return [f"deploy_env '{self.id}': {refusal}"] if refusal else []
 
+    async def _load_delivered_universe(self, env, context: TaskStepContext) -> None:
+        """Load this step's universe when the env's provider promises a loaded env.
+
+        Such a provider (``delivers_universe``) makes deploy_env with an artifact a contract:
+        the env has that universe whether the provider handed over one already holding it or
+        built one cold. Its task carries no load_artifact step, so the load happens here,
+        skipped when the env arrived with this universe untouched, and is recorded the way
+        load_artifact records it. Every other provider leaves loading to load_artifact.
+        """
+        from agent_env.artifact import Artifact, EnvironmentUniverseArtifact
+        from agent_env.env.envs.multi_env import MultiEnv
+        from agent_env.task_step.task_steps.load_artifact import LoadArtifactTaskStep
+
+        if self.artifact_id is None or not isinstance(env, MultiEnv):
+            return
+        if not getattr(getattr(env, "_env_provider", None), "delivers_universe", False):
+            return
+        artifact = Artifact.get(self.artifact_id, self.artifact_version)
+        if not isinstance(artifact, EnvironmentUniverseArtifact):
+            return
+        load_result = await env.load_environment_universe_artifact(artifact, skip_if_preloaded=True)
+        LoadArtifactTaskStep._record_universe_load(context, artifact, load_result)
+
     async def execute(self, context: TaskStepContext) -> TaskStepContext:
         from agent_env.env.env import Env
 
@@ -180,6 +203,7 @@ async def execute(self, context: TaskStepContext) -> TaskStepContext:
         if deployed_env.instance_id:
             update_env_instance_metadata(deployed_env.instance_id, annotations)
         context.deployed_envs.append(deployed_env)
+        await self._load_delivered_universe(env, context)
 
         # Some envs (e.g. harbor coding tasks) build + run the agent as part of
         # deploy() and have no separate DeployAgent step. They advertise the

src/agent_env/task_step/task_steps/load_artifact.py

@@ -267,6 +267,8 @@ def _record_universe_load(context: TaskStepContext, artifact, load_result) -> No
             value = getattr(load_result, key, None)
             if value is not None:
                 entry[key] = value
+        if getattr(load_result, "delivered_preloaded", False):
+            entry["delivered_preloaded"] = True
         context.metadata.setdefault("loaded_environment_universes", []).append(entry)
 
     @staticmethod