diff --git a/examples/sandbox/staging-smoke/probe.py b/examples/sandbox/staging-smoke/probe.py index 3fc0f7180..7e281d39f 100644 --- a/examples/sandbox/staging-smoke/probe.py +++ b/examples/sandbox/staging-smoke/probe.py @@ -184,10 +184,13 @@ async def provision( ) # Reap leftovers this workflow itself may have abandoned (a crashed prior - # run's agent is invisible to any later run once its state file rotates) — - # the toolkit's sweep is prefix- and age-guarded, so nothing else's - # resources are ever touched. - await manager.sweep_orphans() + # run's agent is invisible to any later run once its state file rotates). + # The sweep is prefix- and age-guarded. This run's recorded agent is kept + # by id: its name does not embed the smoke run id, and a resume can outlast + # the age gate. + await manager.sweep_orphans( + keep_ids=[run_state.agent_id] if run_state.agent_id else [] + ) # Re-provisioning within a run (daemon-restart recovery, a crashed # launch): the recorded agent/room belong to a dead process — replace # them. Refuse to stack new resources on top of ones that failed to reap diff --git a/tests/e2e/baseline/README.md b/tests/e2e/baseline/README.md index 420f82bbd..11b066aa0 100644 --- a/tests/e2e/baseline/README.md +++ b/tests/e2e/baseline/README.md @@ -402,7 +402,7 @@ The topology is guarded two ways so a mis-wired test never false-greens: | Run the same scenario across every adapter | `@per_adapter()` → `agent` (id via `agent.adapter_id`) | | Run a scenario across a subset | `@per_adapter(Adapter.X, Adapter.Y)` (positional = include) / `@per_adapter(exclude= by id, supports=/without={Capability.MEMORY} by capability, or runs_tool_loop=True for the custom-tool subgroup)`; add `prompt=`/`features=` (or `**MEMORY_AGENT`) to steer | | Drive the agent lifecycle myself (build-only / reboot / rehydration) | `@per_adapter()` → `cell` (`cell.build()` / `cell.provision()` / `cell.run_as()` — see **AdapterCell**) | -| Clean up what I created | nothing: `resource_manager` reaps on teardown (`BAND_E2E_AUTOCLEAN=false` keeps it for debugging) | +| Clean up what I created | nothing: `resource_manager` reaps on teardown (`BAND_E2E_AUTOCLEAN=false` keeps it for debugging). Each provision gets its own name, so a missed reap cannot block the next cell. | | Drive the platform as a user | the `user_ops` fixture (`UserOps`) | | List who the user could invite to a room | `await user_ops.lookup_peers(not_in_room=room_id)` → `list[Peer]` (the invitable roster; `peer_type="Agent"` narrows) | | Drive a peer agent (e.g. the `Echo` bounce) | `await resource_manager.peer(peer).send_message(room_id, "ECHO: ...", mention_id=agent.id, mention_name=agent.name)` — posts as that agent, returns the message id to barrier on (needs the peer already in the room) | diff --git a/tests/e2e/baseline/fixtures/platform.py b/tests/e2e/baseline/fixtures/platform.py index b28ed69a5..fd7686446 100644 --- a/tests/e2e/baseline/fixtures/platform.py +++ b/tests/e2e/baseline/fixtures/platform.py @@ -50,7 +50,7 @@ def baseline_settings() -> BaselineSettings: @pytest.fixture(scope="session") def baseline_run_id() -> str: - """Token identifying this session's provisioned resources (for naming/sweep).""" + """Token for this session's orphan-sweep self-skip.""" return new_run_id() @@ -113,18 +113,18 @@ async def resource_manager( ) -> AsyncGenerator[ResourceManager, None]: """Per-test provision/reap driver. - Teardown force-deletes everything provisioned this run, unless - ``BAND_E2E_AUTOCLEAN`` is false (kept for on-purpose debugging; surviving - ids are logged by ``reap_all``/``provision_*``). + Teardown force-deletes everything provisioned this run via + ``reap_on_exit``, unless ``BAND_E2E_AUTOCLEAN`` is false (kept for + on-purpose debugging; surviving ids are logged by ``reap_all`` / + ``provision_*``). """ resources = ResourceManager( user_client=baseline_user_client, settings=baseline_settings, run_id=baseline_run_id, ) - yield resources - if baseline_settings.run.autoclean: - await resources.reap_all() + async with resources.reap_on_exit(enabled=baseline_settings.run.autoclean): + yield resources @pytest.fixture(scope="session", autouse=True) diff --git a/tests/e2e/baseline/toolkit/provisioning.py b/tests/e2e/baseline/toolkit/provisioning.py index 0f6385a65..1fadd75c5 100644 --- a/tests/e2e/baseline/toolkit/provisioning.py +++ b/tests/e2e/baseline/toolkit/provisioning.py @@ -53,6 +53,9 @@ # All provisioned agent names start with this; the orphan sweep matches on it. NAME_PREFIX = "e2e-band-" +# Between the token and the label. Both name budgets subtract this same separator. +TOKEN_LABEL_SEPARATOR = "-" + # The platform truncates @mention handles beyond this many characters. A peer that a # scenario provisions and later asserts by @mention (a roster reply names it as # ``@handle``) only surfaces its full name if that name fits the cap; beyond it the @@ -69,29 +72,70 @@ # silently truncating a handle. MAX_MENTIONED_LABEL_LEN = len("invitable") # 9 +# Platform ``validate_length(:name, max: 100)``. A longer register is rejected +# for length, which is a different 422 from name-taken. +AGENT_NAME_MAX = 100 + +# ``uuid4().hex`` width. Labels that may exceed the mention cap take the whole +# uuid so a leftover row cannot collide with the next provision of that label. +NAME_TOKEN_HEX = 32 + + +def name_budget(limit: int, label_len: int) -> int: + """Characters left for the token once the prefix, separator, and label are counted.""" + return limit - len(NAME_PREFIX) - len(TOKEN_LABEL_SEPARATOR) - label_len + def run_id_len() -> int: - """Hex length for a run id that keeps the longest @mentioned peer name within - the handle cap: ``CAP - len(prefix) - len("-") - longest mentioned label``.""" - return MENTION_HANDLE_CAP - len(NAME_PREFIX) - 1 - MAX_MENTIONED_LABEL_LEN + """Hex length of a mentionable provision token. + The longest label a scenario can @mention stays inside the handle cap at + this width. :func:`name_token` uses it for those labels. + """ + return name_budget(MENTION_HANDLE_CAP, MAX_MENTIONED_LABEL_LEN) + + +def normalized_label(label: str) -> str: + return label.replace("_", "-") -def agent_name(*, run_id: str, label: str) -> str: - """The provisioned name for ``label`` in run ``run_id``. + +def agent_name(*, token: str, label: str) -> str: + """The provisioned name for ``label`` and ``token``. Underscores become hyphens: models rewrite ``omp_acp`` as ``omp-acp`` when repeating a name, so an adapter-id label must already be in the form they echo. """ - return f"{NAME_PREFIX}{run_id}-{label.replace('_', '-')}" + return f"{NAME_PREFIX}{token}{TOKEN_LABEL_SEPARATOR}{normalized_label(label)}" + + +def name_token(label: str) -> str: + """A fresh hex token for one provision of ``label``. + + A label a scenario can @mention gets :func:`run_id_len` characters, so the + longest such name stays inside ``MENTION_HANDLE_CAP``. A longer label is not + a mention handle and gets a full uuid slice, clamped so the name stays + inside ``AGENT_NAME_MAX``. A fresh token per provision is what keeps a + leftover row from colliding with the next register of the same label. + """ + normalized = normalized_label(label) + if len(normalized) <= MAX_MENTIONED_LABEL_LEN: + return uuid.uuid4().hex[: run_id_len()] + width = min(NAME_TOKEN_HEX, name_budget(AGENT_NAME_MAX, len(normalized))) + if width < run_id_len(): + raise ValueError( + f"label {normalized!r} leaves {width} characters for a unique name " + f"token; need at least {run_id_len()} within the " + f"{AGENT_NAME_MAX}-character name limit" + ) + return uuid.uuid4().hex[:width] def new_run_id() -> str: - """Short token identifying a single test session's provisioned resources. + """Short token the session fixture draws once for the orphan sweep's self-skip. - Its length is *derived* from the mention-handle cap (see :func:`run_id_len`) so - the longest @mentioned peer name (see :func:`agent_name`) still surfaces in - full rather than as a truncated handle. The resulting entropy - (~1M at 5 hex) is ample given the run-id + age guards in ``sweep_orphans``. + Provisions do not embed it; they draw :func:`name_token`. The sweep skips + names that still contain this token. Per-provision names rely on the age + gate, the same protection another job's agents have. """ return uuid.uuid4().hex[: run_id_len()] @@ -307,7 +351,7 @@ async def provision_agent( (default ``E2E baseline test agent ({label})``). Tests that assert on passive-roster description surfacing pass a self-sourced marker here. """ - name = agent_name(run_id=self._run_id, label=label) + name = agent_name(token=name_token(label), label=label) agent_description = description or f"E2E baseline test agent ({label})" response = await self._client.human_api_agents.register_my_agent( agent=AgentRegisterRequest( @@ -413,13 +457,32 @@ async def reap_all(self) -> None: except Exception: logger.warning("Failed to reap agent %s", agent_id, exc_info=True) - async def sweep_orphans(self) -> int: + @asynccontextmanager + async def reap_on_exit(self, *, enabled: bool = True) -> AsyncGenerator[None, None]: + """Reap everything provisioned in this block when it exits. + + ``enabled`` is false when ``BAND_E2E_AUTOCLEAN`` is off, so a debug run + keeps the resources. ``reap_all`` logs a delete failure and continues, so + a teardown error does not replace an exception from the block. A row + whose id was never returned is not tracked and is not deleted here; the + next provision draws a new name, so that leftover cannot collide. + """ + try: + yield + finally: + if enabled: + await self.reap_all() + + async def sweep_orphans(self, *, keep_ids: Sequence[str] = ()) -> int: """Force-delete stale test agents left by crashed prior runs. - Only touches agents whose name carries ``NAME_PREFIX``, belongs to a - *different* run, and is older than ``orphan_max_age_minutes`` — so a - concurrent run on the same shared platform is never deleted mid-flight. - Returns the number of agents reaped. + Only touches agents whose name carries ``NAME_PREFIX`` and that are + older than ``orphan_max_age_minutes``. Names that still embed this + session's run id are skipped, and so are ``keep_ids``: a per-provision + name does not embed that id, so a caller that still owns an agent (the + staging probe's recorded one, across a resume longer than the age gate) + passes its id here. A concurrent run's fresh agents stay because they + are younger than the age gate. Returns the number of agents reaped. """ max_age = timedelta(minutes=self._settings.run.orphan_max_age_minutes) cutoff = datetime.now(UTC) - max_age @@ -429,6 +492,7 @@ async def sweep_orphans(self) -> int: # Cursor pagination (the SDK's preferred path; offset `page`/`page_size` # is deprecated): advance via metadata.next_cursor until has_more is # false. The iteration cap bounds a best-effort sweep. + owned = set(keep_ids) orphans: list[str] = [] cursor: str | None = None for _ in range(20): @@ -438,8 +502,12 @@ async def sweep_orphans(self) -> int: for agent in response.data or []: if not agent.name.startswith(NAME_PREFIX): continue # name filter is a contains-match; re-check the prefix - if f"-{self._run_id}-" in agent.name: - continue # never reap our own run + if ( + agent.id in owned + or f"{TOKEN_LABEL_SEPARATOR}{self._run_id}{TOKEN_LABEL_SEPARATOR}" + in agent.name + ): + continue # inserted_at may be tz-naive depending on serialization; treat # naive as UTC so the comparison never raises (see the codebase's # _coerce_inserted_at). A naive>aware compare would TypeError and @@ -618,9 +686,9 @@ def build( async def provision(self, *, label: str | None = None) -> ProvisionedAgent: """Register an identity for this cell (tracked + reaped by the manager); no run. - ``label`` defaults to the adapter id (a readable provisioned name). Pass a - distinct label to register more than one identity of the same cell in a single - test, else the generated names collide. + ``label`` defaults to the adapter id (a readable provisioned name). Each + call draws its own name token, so the same label can be registered again + in one test without colliding with an agent this run already created. """ provisioned = await self.resources.provision_agent(label or self.adapter_id) return replace(provisioned, adapter_id=self.adapter_id) @@ -708,7 +776,7 @@ async def run_many( what a same-adapter co-residency gate must probe. ``labels`` default to ``{adapter_id}-{index}``; an explicit list must be length - ``count`` so the provisioned names don't collide. ``prompt`` / ``features`` / + ``count`` (one label per member). ``prompt`` / ``features`` / ``tools`` pass through to each :meth:`run_as`, preserving the cell defaults when omitted. """ diff --git a/tests/framework_conformance/test_provision_names.py b/tests/framework_conformance/test_provision_names.py new file mode 100644 index 000000000..635294699 --- /dev/null +++ b/tests/framework_conformance/test_provision_names.py @@ -0,0 +1,125 @@ +"""Each provision registers under its own agent name. + +A leftover row for a label must not make the next provision of that label +ask the platform for the same name. +""" + +from __future__ import annotations + +from typing import cast +from unittest.mock import AsyncMock, MagicMock + +import pytest +from band_rest import AgentRegisterRequest + +from tests.e2e.baseline.settings import BaselineSettings +from tests.e2e.baseline.toolkit.provisioning import ( + AGENT_NAME_MAX, + MENTION_HANDLE_CAP, + NAME_PREFIX, + NAME_TOKEN_HEX, + ResourceManager, + name_budget, + name_token, + run_id_len, +) + +REGISTERED_AGENT_ID = "registered-agent" + + +def recording_manager() -> tuple[ResourceManager, AsyncMock]: + def respond(*, agent: AgentRegisterRequest) -> MagicMock: + response = MagicMock() + response.data.agent.id = REGISTERED_AGENT_ID + response.data.credentials.api_key = "test-key" + return response + + register = AsyncMock(side_effect=respond) + client = MagicMock() + client.human_api_agents.register_my_agent = register + manager = ResourceManager( + user_client=client, + settings=cast(BaselineSettings, MagicMock()), + run_id="session", + ) + return manager, register + + +def registered_names(register: AsyncMock) -> list[str]: + return [call.kwargs["agent"].name for call in register.await_args_list] + + +def name_parts(name: str) -> tuple[str, str]: + """Hex token and label from a provisioned name. The token contains no hyphen.""" + token, _separator, label = name.removeprefix(NAME_PREFIX).partition("-") + return token, label + + +def is_hex(token: str) -> bool: + return bool(token) and all(character in "0123456789abcdef" for character in token) + + +class TestProvisionNames: + async def test_same_mentioned_label_registers_two_names_inside_the_cap( + self, + ) -> None: + manager, register = recording_manager() + + first = await manager.provision_agent("invitable") + second = await manager.provision_agent("invitable") + + names = registered_names(register) + assert names == [first.name, second.name] + assert first.id == REGISTERED_AGENT_ID + assert first.name != first.id + assert len(set(names)) == 2 + for name in names: + token, label = name_parts(name) + assert name.startswith(NAME_PREFIX) + assert label == "invitable" + assert is_hex(token) + assert len(token) == run_id_len() + assert len(name) == MENTION_HANDLE_CAP + + short = await manager.provision_agent("echo") + short_token, short_label = name_parts(short.name) + assert short_label == "echo" + assert len(short_token) == run_id_len() + assert len(short.name) <= MENTION_HANDLE_CAP + + async def test_adapter_label_uses_a_wide_token_within_the_name_limit( + self, + ) -> None: + manager, register = recording_manager() + + first = await manager.provision_agent("copilot_sdk") + second = await manager.provision_agent("copilot_sdk") + + names = registered_names(register) + assert names == [first.name, second.name] + assert first.id == REGISTERED_AGENT_ID + assert first.name != first.id + assert len(set(names)) == 2 + for name in names: + token, label = name_parts(name) + assert label == "copilot-sdk" + assert is_hex(token) + assert len(token) == 32 + assert len(name) <= AGENT_NAME_MAX + + async def test_long_label_clamps_the_token_to_the_name_limit(self) -> None: + manager, register = recording_manager() + # One shorter than a full uuid, so the token has to shrink to fit. + label = "x" * (name_budget(AGENT_NAME_MAX, 0) - (NAME_TOKEN_HEX - 1)) + + await manager.provision_agent(label) + + token, parsed = name_parts(registered_names(register)[0]) + assert parsed == label + assert len(token) < NAME_TOKEN_HEX + assert len(token) >= run_id_len() + assert len(registered_names(register)[0]) == AGENT_NAME_MAX + + def test_label_too_long_for_a_unique_name_is_rejected(self) -> None: + with pytest.raises(ValueError, match="unique name token"): + name_token("x" * 90) diff --git a/tests/framework_conformance/test_reap_on_exit.py b/tests/framework_conformance/test_reap_on_exit.py new file mode 100644 index 000000000..f8b9a870c --- /dev/null +++ b/tests/framework_conformance/test_reap_on_exit.py @@ -0,0 +1,63 @@ +"""Test-end cleanup reaps tracked resources without hiding the test error.""" + +from __future__ import annotations + +from typing import cast +from unittest.mock import AsyncMock, MagicMock + +import pytest + +from tests.e2e.baseline.settings import BaselineSettings +from tests.e2e.baseline.toolkit.provisioning import ResourceManager + +REGISTERED_AGENT_ID = "agent-1" + + +def manager() -> tuple[ResourceManager, AsyncMock]: + def respond(**_kwargs: object) -> MagicMock: + response = MagicMock() + response.data.agent.id = REGISTERED_AGENT_ID + response.data.credentials.api_key = "test-key" + return response + + client = MagicMock() + client.human_api_agents.register_my_agent = AsyncMock(side_effect=respond) + delete = AsyncMock() + client.human_api_agents.delete_my_agent = delete + return ( + ResourceManager( + user_client=client, + settings=cast(BaselineSettings, MagicMock()), + run_id="session", + ), + delete, + ) + + +class TestReapOnExit: + async def test_reaps_on_normal_exit(self) -> None: + resources, delete = manager() + + async with resources.reap_on_exit(): + await resources.provision_agent("copilot_sdk") + + delete.assert_awaited_once_with(REGISTERED_AGENT_ID, force=True) + + async def test_body_error_survives_a_failed_reap(self) -> None: + resources, delete = manager() + delete.side_effect = RuntimeError("delete failed") + + with pytest.raises(RuntimeError, match="boom"): + async with resources.reap_on_exit(): + await resources.provision_agent("copilot_sdk") + raise RuntimeError("boom") + + delete.assert_awaited_once_with(REGISTERED_AGENT_ID, force=True) + + async def test_disabled_does_not_reap(self) -> None: + resources, delete = manager() + + async with resources.reap_on_exit(enabled=False): + await resources.provision_agent("copilot_sdk") + + delete.assert_not_awaited() diff --git a/tests/framework_conformance/test_sweep_owned.py b/tests/framework_conformance/test_sweep_owned.py new file mode 100644 index 000000000..41bba9797 --- /dev/null +++ b/tests/framework_conformance/test_sweep_owned.py @@ -0,0 +1,58 @@ +"""An agent this run still owns is not an orphan, even after the age gate.""" + +from __future__ import annotations + +from datetime import UTC, datetime, timedelta +from typing import cast +from unittest.mock import AsyncMock, MagicMock + +from tests.e2e.baseline.settings import BaselineSettings +from tests.e2e.baseline.toolkit.provisioning import ( + ResourceManager, + agent_name, +) + +OWNED_AGENT_ID = "owned-1" +SESSION_RUN_ID = "78228ce0" + + +def listing_manager() -> tuple[ResourceManager, MagicMock, AsyncMock]: + agent = MagicMock() + agent.id = OWNED_AGENT_ID + agent.name = agent_name(token="abcde", label="sandbox") + agent.inserted_at = datetime.now(UTC) - timedelta(minutes=121) + listed = MagicMock() + listed.data = [agent] + listed.metadata.next_cursor = None + listed.metadata.has_more = False + settings = MagicMock() + settings.run.orphan_max_age_minutes = 120 + client = MagicMock() + client.human_api_agents.list_my_agents = AsyncMock(return_value=listed) + delete = AsyncMock() + client.human_api_agents.delete_my_agent = delete + manager = ResourceManager( + user_client=client, + settings=cast(BaselineSettings, settings), + run_id=SESSION_RUN_ID, + ) + return manager, agent, delete + + +class TestSweepOwned: + async def test_aged_owned_agent_is_not_swept(self) -> None: + manager, agent, delete = listing_manager() + assert f"-{SESSION_RUN_ID}-" not in agent.name + + reaped = await manager.sweep_orphans(keep_ids=[agent.id]) + + assert reaped == 0 + delete.assert_not_awaited() + + async def test_aged_agent_without_an_owner_is_swept(self) -> None: + manager, agent, delete = listing_manager() + + reaped = await manager.sweep_orphans() + + assert reaped == 1 + delete.assert_awaited_once_with(agent.id, force=True) diff --git a/tests/framework_conformance/test_toolkit_helpers.py b/tests/framework_conformance/test_toolkit_helpers.py index c411b790b..4ddfbf826 100644 --- a/tests/framework_conformance/test_toolkit_helpers.py +++ b/tests/framework_conformance/test_toolkit_helpers.py @@ -292,7 +292,7 @@ async def test_adopt_room_is_reaped_once_on_teardown() -> None: def test_adapter_id_labels_become_names_models_echo_verbatim() -> None: """A model repeating ``omp_acp`` writes ``omp-acp``; the name must already match.""" - assert agent_name(run_id="ab12c", label="omp_acp") == "e2e-band-ab12c-omp-acp" + assert agent_name(token="ab12c", label="omp_acp") == "e2e-band-ab12c-omp-acp" def test_mentioned_peer_names_fit_handle_cap() -> None: @@ -315,7 +315,7 @@ def test_mentioned_peer_names_fit_handle_cap() -> None: # The longest label a scenario provisions AND asserts by @mention must fit. longest = "x" * MAX_MENTIONED_LABEL_LEN - name = agent_name(run_id="a" * length, label=longest) + name = agent_name(token="a" * length, label=longest) assert len(name) <= MENTION_HANDLE_CAP, ( f"longest @mentioned peer name {name!r} is {len(name)} chars, over the " f"{MENTION_HANDLE_CAP}-char mention-handle cap"