feat(task): work out what a stored task requires by the rule bundle register uses

AGENTS.md

@@ -196,8 +196,11 @@ Out-of-tree code plugs in either through `config.toml`, as a `module:Class` poin
   met, or whose steps read an env or artifact of a type nothing here registers, offering in a terminal to install what
   it requires (`cli/task/_checks.py`), then set up in local stores the infra envs its gateway deploys run on
   (`preflight.stored_infra`). A step type nothing registers names the task's unmet requirements in the load error
-  (`UnknownStepTypeError`). A task stored without `requires` loads and runs as before; a hub edit drops it until the
-  hub passes it through.
+  (`UnknownStepTypeError`). A task stored without `requires` loads and runs as before. A service that saves tasks
+  computes it by the same rule with `task_requires(steps, declared)` (`bundle/requires.py`), reading the envs, agents
+  and artifacts the steps reference, and those they hold, from the store: a package the previous version required
+  keeps its range, one used anew is at least the installed version, and one no longer used is dropped unless a
+  reference can't be read.
 - Environment providers have no table: an installed package registers one as an `agent_env.env_providers`
   entry point named its `type`, which records carry as `env_provider_type`.
 - `[plugins.<distribution name>]` is reserved for an installed plugin's own settings, read with

src/agent_env/bundle/requires.py

@@ -8,7 +8,7 @@
 import json
 import os
 import uuid
-from collections.abc import Iterable, Mapping, Sequence
+from collections.abc import Iterable, Iterator, Mapping, Sequence
 from dataclasses import dataclass
 from enum import Enum
 from importlib.metadata import distribution, packages_distributions
@@ -19,12 +19,19 @@
 from packaging.utils import canonicalize_name
 from packaging.version import InvalidVersion, Version
 
+from agent_env.a2a_agent.a2a_agent import A2AAgent
+from agent_env.artifact.artifact import Artifact
 from agent_env.artifact.registry import ARTIFACT_REGISTRY, canonical_type, get_artifact_registry
+from agent_env.entity.refs import EntityKind, RefRole, ref_sites
 from agent_env.env import registry as env_registry
+from agent_env.env.env import Env
 from agent_env.env.registry import get_env_registry
 from agent_env.plugins import ARTIFACTS, ENVS, TASK_STEPS, _requirements, inventory, load_failures
+from agent_env.store.base import NotFoundError
+from agent_env.store.document_store import UnknownDocumentTypeError
 from agent_env.task_step import registry as task_step_registry
 from agent_env.task_step.registry import get_task_step_registry
+from agent_env.task_step.task_step import TaskStep
 
 from . import _toml
 from ._fs import relative
@@ -36,6 +43,8 @@
 # What core ships, per plugin group: anything else a bundle uses is provided by a plugin, or by configuration.
 _BUILTINS = {TASK_STEPS: task_step_registry._builtin_registry, ENVS: env_registry._builtin_registry,
              ARTIFACTS: lambda: ARTIFACT_REGISTRY}
+# The entities a step reads that hold, or are, a type a plugin may register.
+_GETTERS = {EntityKind.ENV: Env.get, EntityKind.AGENT: A2AAgent.get, EntityKind.ARTIFACT: Artifact.get}
 
 
 class UnmetRequirements(BundleError):
@@ -180,6 +189,67 @@ def detected(found: Iterable[Use], declared: Iterable[str] = ()) -> list[str]:
     return sorted(requires)
 
 
+def task_requires(steps: Iterable[TaskStep], declared: Iterable[str] = ()) -> list[str]:
+    """What a stored task made of ``steps`` requires, by the rule registering a bundle's task records: the plugin
+    packages behind its own step types and the types of the envs and artifacts its steps read, as the store holds them,
+    with those they and the agents its steps deploy hold. A package ``declared`` (what the task required before) names keeps its range there; one used
+    anew is at least the version installed here; one no longer used is dropped, unless a reference can't be read here
+    (an id the run fills in, one the store doesn't hold, a type nothing here registers), when every one is kept."""
+    declared = list(declared)
+    providers = Providers()
+    found, read, unread = [], set(), False
+    for step in steps:
+        if (provider := providers.of(TASK_STEPS, step.type, get_task_step_registry().get(step.type))) is not None:
+            found.append(Use(f"step {step.id!r}", "step type", step.type, provider))
+        for site in ref_sites(type(step).entity_refs or (), step.to_dict()):
+            if site.ref.role is RefRole.OUTPUT or site.ref.kind not in _GETTERS:
+                continue
+            key = (site.ref.kind, site.value, site.version if isinstance(site.version, int) else None)
+            if key in read:
+                continue
+            read.add(key)
+            try:
+                entity = _GETTERS[site.ref.kind](key[1], key[2]) if isinstance(key[1], str) else None
+            except (NotFoundError, UnknownDocumentTypeError, ValueError):
+                entity = None
+            if entity is None:
+                unread = True
+                continue
+            found += [use for held in _held(entity) if (use := _entity_use(f"step {step.id!r}: {site.path}", held,
+                                                                           providers)) is not None]
+    requires = detected(found, declared)
+    if unread:
+        requires = sorted({*requires, *(line for line in declared if _requirements.parsed(line) is not None)})
+    return requires
+
+
+
+def _held(entity: Env | A2AAgent | Artifact) -> Iterator[Env | A2AAgent | Artifact]:
+    """``entity``, then each env and artifact loading it loaded with it, and theirs in turn."""
+    seen, queue = set(), [entity]
+    while queue:
+        held = queue.pop()
+        if id(held) in seen:
+            continue
+        seen.add(id(held))
+        yield held
+        for value in vars(held).values():
+            items = value.values() if isinstance(value, dict) else value if isinstance(value, (list, tuple)) else (value,)
+            queue.extend(item for item in items if isinstance(item, (Env, Artifact)))
+
+
+def _entity_use(where: str, entity: Env | A2AAgent | Artifact, providers: Providers) -> Use | None:
+    """The use of ``entity``'s type, when core doesn't ship it; an agent's has no plugin group."""
+    if isinstance(entity, Env):
+        provider = providers.of(ENVS, entity.type, get_env_registry().get(entity.type))
+        return Use(where, "env type", entity.type, provider) if provider is not None else None
+    if not isinstance(entity, Artifact):
+        return None
+    name = canonical_type(entity.type)
+    provider = providers.of(ARTIFACTS, name, get_artifact_registry().get(name))
+    return Use(where, "artifact type", name, provider) if provider is not None else None
+
+
 def declare(resolved: ResolvedBundle, declaring: Declaring) -> list[str]:
     """What to say of the plugin packages the bundle uses that its ``bundle.toml`` doesn't require, and of each type
     only this machine's configuration registers, as ``declaring`` says. Writing, a bundle with no ``bundle.toml`` is

tst/unit/bundle/requires_test.py

@@ -15,24 +15,30 @@
 from click.testing import CliRunner
 from packaging.requirements import Requirement
 
+from agent_env.a2a_agent import A2AAgent
 from agent_env.artifact import Artifact
+from agent_env.artifact.artifacts.docker_image import DockerImageArtifact
+from agent_env.artifact.store import get_artifact_store
 from agent_env.bundle import BundleError, dry_run_bundle, parse_bundle, prepare_bundle, run_bundle
 from agent_env.bundle import register as register_module
 from agent_env.bundle import requires as requires_module
 from agent_env.bundle.importer import import_task
 from agent_env.bundle.register import register_bundle
 from agent_env.bundle.requires import (
-    Declaring, Provider, Providers, UnmetRequirements, _create, declare, declaring_for,
+    Declaring, Provider, Providers, UnmetRequirements, _create, declare, declaring_for, task_requires,
 )
 from agent_env.bundle.resolve import resolve_bundle
 from agent_env.bundle.run import _write_bundle
 from agent_env.cli import _install_prompt, cli
 from agent_env.cli._installers import InstallerError
+from agent_env.config import get_config
 from agent_env.config.runtime import Config
+from agent_env.env import MCPServerEnv
 from agent_env.plugins import ARTIFACTS, TASK_STEPS, Contribution, Distribution
 from agent_env.plugins._requirements import install_command, unmet, unmet_problems
 from agent_env.store.routing import namespace_routing
 from agent_env.task import Task
+from agent_env.task.store import TASKS_COLLECTION
 from agent_env.task_step.task_step import TaskStep
 from tst.unit._terminal import at_a_terminal
 from tst.unit.bundle._support import fake_plugins, layout
@@ -612,19 +618,33 @@ def from_toml(cls, data, ctx):
 }
 
 
-class TestRegisteredTasks:
-    @pytest.fixture(autouse=True)
-    def two_plugins(self, stores, monkeypatch):
-        """agentenv-acme's step type, and agentenv-acme-settings's artifact type, which a task references."""
-        monkeypatch.setattr(register_module, "_git", lambda root, *args: None)
-        artifacts = {**Config().artifact_registry(), _AcmeSetting.model_fields["type"].default: _AcmeSetting}
-        monkeypatch.setattr(Config, "artifact_registry", lambda self: artifacts)
-        fake_plugins(monkeypatch, [
-            Distribution(ACME, "1.2.0", (Contribution(TASK_STEPS, _Acme.type, _value(_Acme), "active"),)),
-            Distribution(SETTINGS, "2.0", (Contribution(ARTIFACTS, _AcmeSetting.model_fields["type"].default,
-                                                         _value(_AcmeSetting), "active"),)),
-        ], versions={ACME: "1.2.0", SETTINGS: "2.0"})
+class _AcmeImage(DockerImageArtifact):
+    """An image artifact type the plugin agentenv-acme-images ships, which an env holds."""
+
+    type: Literal["acme_image_requires_test"] = "acme_image_requires_test"
+
+
+IMAGES = "agentenv-acme-images"
 
+
+@pytest.fixture
+def plugins(stores, monkeypatch):
+    """agentenv-acme's step type, agentenv-acme-settings's artifact type, which a task references, and
+    agentenv-acme-images's, which an env holds."""
+    monkeypatch.setattr(register_module, "_git", lambda root, *args: None)
+    artifacts = {**Config().artifact_registry(),
+                 **{cls.model_fields["type"].default: cls for cls in (_AcmeSetting, _AcmeImage)}}
+    monkeypatch.setattr(Config, "artifact_registry", lambda self: artifacts)
+    fake_plugins(monkeypatch, [
+        Distribution(ACME, "1.2.0", (Contribution(TASK_STEPS, _Acme.type, _value(_Acme), "active"),)),
+        *(Distribution(name, version, (Contribution(ARTIFACTS, cls.model_fields["type"].default, _value(cls),
+                                                    "active"),))
+          for name, version, cls in ((SETTINGS, "2.0", _AcmeSetting), (IMAGES, "3.0", _AcmeImage))),
+    ], versions={ACME: "1.2.0", SETTINGS: "2.0", IMAGES: "3.0"})
+
+
+@pytest.mark.usefixtures("plugins")
+class TestRegisteredTasks:
     def test_each_task_requires_what_it_and_the_bundle_entities_it_references_use(self, stores):
         root = layout(stores / "b", REGISTERED)
 
@@ -659,5 +679,55 @@ def test_an_import_of_a_registered_task_keeps_the_range_it_records(self, stores)
         assert parse_bundle(imported.root).requires == ("agentenv-acme>=1.0,<2",)
 
 
+def _steps(steps):
+    """The steps of task ``t``, stored as ``steps`` document by document, as ``Task.get`` loads them."""
+    get_config().get_document_store().insert(TASKS_COLLECTION, {"id": "t", "type": "task", "version": 1,
+                                                                "steps": steps})
+    return Task.get("t").steps
+
+
+@pytest.mark.usefixtures("plugins")
+class TestStoredTasks:
+    def test_a_registered_task_requires_what_registering_it_recorded(self, stores):
+        register_bundle(layout(stores / "b", REGISTERED))
+
+        for name in ("plugin", "loads", "core"):
+            task = Task.get(f"b-{name}")
+            assert task_requires(task.steps) == (task.requires or [])
+
+    def test_a_package_it_required_keeps_its_range_one_used_anew_is_at_least_the_installed_version(self, stores):
+        register_bundle(layout(stores / "b", REGISTERED))
+        declared = ["agentenv-acme>=1.0,<2", "agentenv-acme-settings"]
+
+        assert task_requires(Task.get("b-plugin").steps, declared) == ["agentenv-acme>=1.0,<2"]
+        assert task_requires(Task.get("b-loads").steps, ["agentenv-acme>=1.0,<2"]) == ["agentenv-acme-settings>=2.0"]
+        assert task_requires(Task.get("b-core").steps, declared) == []
+
+    def test_a_type_only_configuration_registers_is_never_required(self, stores):
+        assert task_requires(_steps([{"id": "s", "type": _Configured.type}])) == []
+
+    @pytest.mark.parametrize("step", [{"id": "env", "type": "deploy_env", "env_id": "crm"},
+                                      {"id": "agent", "type": "deploy_agent", "a2a_agent_id": "coder"}],
+                             ids=["env", "agent"])
+    def test_a_type_an_env_or_agent_it_reads_holds_is_required(self, stores, step):
+        with namespace_routing():
+            image = get_artifact_store().put_document(_AcmeImage(
+                id="acme-image", description="acme", image_name="acme:latest", tar_gz_s3_url=None, platform=None))
+            MCPServerEnv.put(id="crm", docker_image_artifact=image, environment_name="crm")
+            A2AAgent.put(id="coder", docker_image_artifact=image)
+
+        assert task_requires(_steps([step])) == ["agentenv-acme-images>=3.0"]
+
+    @pytest.mark.parametrize("artifact_id", ["absent", "{{seed.artifact}}", "unknown"])
+    def test_a_reference_it_cant_read_keeps_every_package_it_required(self, stores, artifact_id):
+        get_config().get_document_store().insert("artifacts", {"id": "unknown", "type": "acme_unknown_requires_test",
+                                                               "version": 1})
+        steps = _steps([SANDBOX, {"id": "s", "type": _Acme.type},
+                        {"id": "load", "type": "load_artifact", "sandbox_name": "box", "artifact_id": artifact_id,
+                         "depends_on": [{"task_step_id": "box"}]}])
+
+        assert task_requires(steps, ["agentenv-acme-settings", "agentenv-acme>=1.0,<2", "not a requirement!"]) == [
+            "agentenv-acme-settings", "agentenv-acme>=1.0,<2"]
+
 def test_unmet_requirements_are_a_bundle_error():
     assert issubclass(UnmetRequirements, BundleError)