langflow/scripts/e2e_deployment_tests/watsonx_orchestrate/adapter.py

"""Direct Watsonx adapter scenario runner.

Runs scenario matrices against `WatsonxOrchestrateDeploymentService` directly
(no `/api/v1/deployments` calls).

Warning:
--------
This script performs live integration calls and creates real resources in
Watsonx Orchestrate (agents, snapshots/tools, and configs/connections).
By default, cleanup runs at the end of execution, but cleanup is best-effort:
if the process is interrupted or provider deletes fail, resources may remain.
Use `--keep-resources` only when you intentionally want to inspect leftovers.

Scenario catalog
----------------
Live create scenarios:
- `live_create_success`: creates config + snapshot + agent successfully (expects Success).
- `live_invalid_config_reference`: rejects create payload when provider_data references
  a non-existent existing connection id (expects InvalidContentError).
- `live_duplicate_snapshot_names_conflict`: duplicate raw snapshot names are deduped
  by schema normalization (expects Success).

Live lifecycle scenarios:
- `live_lifecycle_create_seed`: creates a seed deployment for lifecycle checks (expects Success).
- `live_list_contains_seed`: verifies list-by-id includes the seed deployment (expects Success).
- `live_get_seed`: fetches deployment details by id (expects Success).
- `live_update_seed_name_description`: updates deployment name/description (expects Success).
- `live_get_after_update_reflects_name`: confirms updated name is persisted (expects Success).
- `live_get_status_connected`: confirms status endpoint reports connected deployment (expects Success).
- `live_create_execution_success`: starts an execution run with valid message payload and validates
  provider_result contains execution_id (not run_id) and status (expects Success).
- `live_create_execution_input_string`: creates execution using plain string input (expects Success).
- `live_create_execution_input_dict_content`: creates execution using dict input with content key (expects Success).
- `live_get_execution_poll_terminal`: polls execution until terminal status and validates
  provider_result fields including execution_id, status, and absence of run_id (expects Success).
- `live_get_execution_terminal_fields`: validates terminal execution has agent_id and
  appropriate timestamp (completed_at, failed_at, or cancelled_at) (expects Success).
- `live_delete_seed`: deletes seed deployment agent (expects Success).
- `live_get_after_delete_not_found`: confirms deleted deployment is no longer fetchable
  (expects DeploymentNotFoundError).
- `live_status_after_delete_not_found_state`: confirms status on deleted deployment returns not found
  (expects DeploymentNotFoundError).

Live snapshot/config listing scenarios:
- `live_list_snapshots_by_ids_returns_known`: fetches known snapshot ids via
  snapshot_ids mode and confirms all are returned (expects Success).
- `live_list_snapshots_by_ids_filters_unknown`: mixes a known id with a bogus id;
  confirms the provider returns only existing snapshots (expects Success).
- `live_list_snapshots_by_ids_empty_input`: passes an empty list; this normalizes to
  tenant-scoped snapshot listing and should still succeed (expects Success).
- `live_list_snapshots_tenant_scope`: lists tenant-scoped snapshots (expects Success).
- `live_list_configs_tenant_scope`: lists tenant-scoped configs (expects Success).
- `live_list_snapshots_by_names_returns_known`: queries snapshot_names mode and
  confirms known names resolve to known snapshot ids (expects Success).
- `live_list_snapshots_by_names_ignored_with_deployment_scope`: passes both deployment_ids
  and snapshot_names, and confirms deployment scope takes precedence (expects Success).

Live negative scenarios:
- `live_negative_create_seed`: creates a second seed deployment for negative-path checks (expects Success).
- `live_create_execution_rejects_empty_input`: rejects empty execution input payload (expects InvalidContentError).
- `live_create_execution_missing_deployment`: rejects execution for non-existent deployment
  (expects DeploymentNotFoundError).
- `live_delete_missing_not_found`: delete on unknown deployment id returns not found (expects DeploymentNotFoundError).
- `live_negative_delete_seed`: cleans up negative-path seed deployment (expects Success).

Live service-surface scenarios:
- `live_list_types_supports_agent`: lists supported deployment types and validates AGENT is present (expects Success).
- `live_list_llms_returns_models`: lists provider models and validates the normalized payload
  is non-empty (expects Success).
- `live_verify_credentials_success`: verifies configured credentials against the provider instance (expects Success).
- `live_update_snapshot_success`: updates an existing snapshot artifact by id (expects Success).
- `live_rollback_create_result_cleans_up_created`: runs create rollback cleanup using a real create result
  and verifies the created deployment is removed (expects Success).
- `live_redeploy_not_supported`: ensures redeploy returns operation-not-supported semantics
  (expects InvalidDeploymentOperationError).
- `live_duplicate_not_supported`: ensures duplicate returns operation-not-supported semantics
  (expects InvalidDeploymentOperationError).
- `live_teardown_noop`: calls adapter teardown and expects a successful no-op (expects Success).

Live update-matrix scenarios:
- Contract note: in provider_data operations, `app_ids` are operation ids.
  Raw connection `app_id` values are preserved exactly as declared.
- `upd_spec_only_name_desc`: updates deployment metadata only (expects Success).
- `upd_snapshot_remove_only_no_config`: removes an attached snapshot via provider_data operation
  (expects Success).
- `upd_config_only_existing_tools_with_config_id`: rebinds existing attached snapshots
  to an explicit existing app id via provider_data (expects Success).
- `upd_snapshot_add_ids_with_config_id`: binds existing snapshot ids using provider_data
  operations (expects Success).
- `upd_snapshot_add_raw_with_config_id`: creates and binds raw snapshot payloads using
  provider_data tools/connections/operations (expects Success).
- `upd_mixed_add_remove_raw_with_config`: mixed add/remove/raw snapshot update with provider_data
  operations (expects Success).
- `upd_reject_bind_with_undeclared_app_id`: rejects bind operation with undeclared app id
  in provider_data (expects InvalidContentError).
- `upd_reject_raw_bind_with_undeclared_app_id`: rejects raw-tool bind operation with undeclared app id
  in provider_data (expects InvalidContentError).
- `upd_reject_unbind_with_undeclared_app_id`: rejects unbind operation with undeclared app id
  in provider_data (expects InvalidContentError).
- `upd_reject_unbind_unknown_tool_id`: rejects unbind on unknown tool id
  in provider_data (expects InvalidContentError).
- `upd_missing_add_id_fails`: rejects unknown bind.tool.tool_id_with_ref in provider_data (expects InvalidContentError).
- `upd_config_raw_payload_conflict`: detects conflict when creating duplicate provider_data
  raw connection app id (expects ResourceConflictError).
- `upd_not_found_deployment`: update unknown deployment id returns not found (expects DeploymentNotFoundError).
- `upd_put_tools_replaces_tool_list`: uses put_tools to declaratively replace
  the agent's tool list with a subset (expects Success).
- `upd_put_tools_empty_clears_all_tools`: passes an empty put_tools list to
  remove all tools from the agent (expects Success).
- `upd_put_tools_deduplicates`: passes duplicate tool ids in put_tools and
  confirms deduplication produces a single binding (expects Success).

Failpoint scenarios:
- `fp_retry_create_config_then_success`: injects transient config-create failures; retries
  then succeeds (expects Success).
- `fp_non_retryable_create_agent_conflict`: injects non-retryable agent conflict (expects ResourceConflictError).
- `fp_create_agent_failure_triggers_rollback`: injects repeated agent-create failure and
  checks rollback (expects DeploymentError).
- `fp_update_bindings_failure_triggers_rollback`: injects update-stage binding failure
  and validates rollback path (expects DeploymentError).
- `fp_update_bindings_failure_with_rollback_failure`: injects update failure + rollback
  failure and expects terminal error (expects DeploymentError).
- `fp_update_failure_then_put_tools_restore`: injects a binding-stage failure to corrupt
  the tool list, then uses put_tools to restore the original snapshot set and confirms
  the tool list matches the pre-failure state (expects Success after restore).
"""

from __future__ import annotations

import argparse
import asyncio
import os
import re
import textwrap
import types
from contextlib import suppress
from dataclasses import dataclass
from datetime import datetime, timezone
from types import MethodType, SimpleNamespace
from typing import TYPE_CHECKING, Any
from uuid import uuid4

import langflow.services.adapters.deployment.watsonx_orchestrate.core.create as create_core_module
import langflow.services.adapters.deployment.watsonx_orchestrate.core.retry as retry_module
import langflow.services.adapters.deployment.watsonx_orchestrate.core.shared as shared_core_module
import langflow.services.adapters.deployment.watsonx_orchestrate.core.update as update_core_module
from dotenv import load_dotenv
from fastapi import HTTPException
from ibm_watsonx_orchestrate_clients.tools.tool_client import ClientAPIException
from langflow.services.adapters.deployment.context import (
    DeploymentAdapterContext,
    DeploymentProviderIDContext,
)
from langflow.services.adapters.deployment.watsonx_orchestrate import (
    WatsonxOrchestrateDeploymentService,
    WxOCredentials,
)
from langflow.services.adapters.deployment.watsonx_orchestrate.payloads import (
    WatsonxDeploymentCreateResultData,
    WatsonxFlowArtifactProviderData,
)
from lfx.services.adapters.deployment.exceptions import (
    DeploymentError,
    DeploymentNotFoundError,
    InvalidContentError,
    InvalidDeploymentOperationError,
    InvalidDeploymentTypeError,
    OperationNotSupportedError,
    ResourceConflictError,
)
from lfx.services.adapters.deployment.schema import (
    BaseDeploymentData,
    BaseDeploymentDataUpdate,
    BaseFlowArtifact,
    ConfigListParams,
    DeploymentConfig,
    DeploymentCreate,
    DeploymentListParams,
    DeploymentType,
    DeploymentUpdate,
    ExecutionCreate,
    SnapshotListParams,
    VerifyCredentials,
)

OUTCOME_SUCCESS = "Success"
OUTCOME_INVALID_OPERATION = "InvalidDeploymentOperationError"
OUTCOME_CONFLICT = "ResourceConflictError"
OUTCOME_INVALID_CONTENT = "InvalidContentError"
OUTCOME_FAILURE = "DeploymentError"
OUTCOME_NOT_FOUND = "DeploymentNotFoundError"
HTTP_STATUS_NOT_FOUND = 404
HTTP_STATUS_CONFLICT = 409
MIN_MIXED_SNAPSHOT_IDS = 2
DEFAULT_CONCURRENCY_ITERATIONS = 1
EXECUTION_POLL_INTERVAL_SECS = 2
EXECUTION_POLL_MAX_ATTEMPTS = 10
EXECUTION_TERMINAL_STATUSES = {"completed", "failed", "cancelled", "async_completed", "expired", "requires_input"}
DEFAULT_WXO_LLM = "groq/openai/gpt-oss-120b"

_INVALID_WXO_NAME_CHARS = re.compile(r"[^A-Za-z0-9_]")

if TYPE_CHECKING:
    from collections.abc import Awaitable, Callable


class DummySettingsService:
    def __init__(self) -> None:
        self.settings = SimpleNamespace()


@dataclass(slots=True)
class ScenarioResult:
    name: str
    expected_outcomes: set[str]
    actual_outcome: str
    ok: bool
    detail: str


@dataclass(slots=True)
class WxoCreatedDeploymentResult:
    deployment_id: str
    provider_result: WatsonxDeploymentCreateResultData


class WatsonxAdapterDirectE2E:
    def __init__(
        self,
        *,
        provider_backend_url: str,
        provider_api_key: str,
        project_id: str,
        mode: str,
        keep_resources: bool,
        llm: str,
    ) -> None:
        self.provider_backend_url = provider_backend_url
        self.provider_api_key = provider_api_key
        self.project_id = project_id
        self.mode = mode
        self.keep_resources = keep_resources
        self.llm = llm
        self.run_suffix = datetime.now(timezone.utc).strftime("%Y%m%d%H%M%S") + "-" + uuid4().hex[:8]

        self.user_id = str(uuid4())
        self.db = object()
        self.provider_id = uuid4()
        self.service = WatsonxOrchestrateDeploymentService(DummySettingsService())

        import langflow.services.adapters.deployment.watsonx_orchestrate.client as _client_mod

        self._client_mod = _client_mod

        deployment_context = DeploymentAdapterContext(provider_id=self.provider_id)
        self._deployment_context_token = DeploymentProviderIDContext.set_current(deployment_context)

        self._original_resolve_wxo_client_credentials = _client_mod.resolve_wxo_client_credentials

        async def _resolve_credentials(*, user_id, db, provider_id):  # noqa: ARG001
            authenticator = _client_mod.get_authenticator(
                instance_url=self.provider_backend_url,
                api_key=self.provider_api_key,
            )
            return WxOCredentials(instance_url=self.provider_backend_url, authenticator=authenticator)

        _client_mod.resolve_wxo_client_credentials = _resolve_credentials  # type: ignore[assignment]

        self.created_deployment_ids: set[str] = set()
        self.created_snapshot_ids: set[str] = set()
        self.created_config_ids: set[str] = set()
        self.tool_source_refs: dict[str, str] = {}

    async def run(self) -> int:
        print("Starting watsonx direct adapter runner...")
        print(f"mode={self.mode} project_id={self.project_id} keep_resources={self.keep_resources} llm={self.llm}")
        try:
            results: list[ScenarioResult] = []
            if self.mode in {"live", "both"}:
                results.extend(await self._run_live_scenarios())
            if self.mode in {"failpoint", "both"}:
                results.extend(await self._run_failpoint_scenarios())

            self._print_summary(results)
            if not self.keep_resources:
                await self._cleanup_resources()
            return 1 if any(not result.ok for result in results) else 0
        finally:
            self._client_mod.resolve_wxo_client_credentials = self._original_resolve_wxo_client_credentials
            self._client_mod.clear_provider_clients_request_context()
            DeploymentProviderIDContext.reset_current(self._deployment_context_token)

    async def _run_live_scenarios(self) -> list[ScenarioResult]:
        duplicate_name = self._mk_name("dup_snapshot")
        scenarios = [
            {
                "name": "live_create_success",
                "expected": {OUTCOME_SUCCESS},
                "payload": self._build_create_payload(
                    tool_payloads=[self._build_flow_payload(label="snap_live_success")],
                    raw_connection=self._build_config_payload(label="cfg_live_success"),
                ),
            },
            {
                "name": "live_invalid_config_reference",
                "expected": {OUTCOME_INVALID_CONTENT},
                "payload": self._build_create_payload(
                    tool_payloads=[self._build_flow_payload(label="snap_live_invalid_ref")],
                    existing_connection_app_id="cfg_ref_not_supported",
                ),
            },
            {
                "name": "live_duplicate_snapshot_names_conflict",
                "expected": {OUTCOME_SUCCESS},
                "payload": self._build_create_payload(
                    tool_payloads=[
                        self._build_flow_payload(label="snap_dup_a", name_override=duplicate_name),
                        self._build_flow_payload(label="snap_dup_b", name_override=duplicate_name),
                    ],
                    raw_connection=self._build_config_payload(label="cfg_live_dup"),
                ),
                "assert_dedupe_snapshot_count": 1,
            },
        ]
        results = await self._run_scenarios(scenarios)
        results.extend(await self._run_live_lifecycle_scenarios())
        results.extend(await self._run_live_list_snapshots_by_ids_scenarios())
        results.extend(await self._run_live_listing_mode_scenarios())
        results.extend(await self._run_live_service_surface_scenarios())
        results.extend(await self._run_live_update_matrix_scenarios())
        results.extend(await self._run_live_concurrency_scenarios())
        results.extend(await self._run_live_negative_scenarios())
        return results

    async def _run_failpoint_scenarios(self) -> list[ScenarioResult]:
        scenarios = [
            {
                "name": "fp_retry_create_config_then_success",
                "expected": {OUTCOME_SUCCESS},
                "payload": self._build_create_payload(
                    tool_payloads=[self._build_flow_payload(label="snap_fp_retry")],
                    raw_connection=self._build_config_payload(label="cfg_fp_retry"),
                ),
                "inject": {
                    "create_config": {"fail_first_n": 2, "error_type": "runtime", "message": "fp_create_config_retry"}
                },
            },
            {
                "name": "fp_create_config_wrapper_conflict_mapping",
                "expected": {OUTCOME_CONFLICT},
                "payload": self._build_create_payload(
                    tool_payloads=[self._build_flow_payload(label="snap_fp_cfg_wrapper_conflict")],
                    raw_connection=self._build_config_payload(label="cfg_fp_cfg_wrapper_conflict"),
                ),
                "inject": {
                    "create_config_wrapper": {
                        "fail_first_n": 1,
                        "error_type": "domain_conflict",
                        "message": "fp_create_config_wrapper_conflict",
                    }
                },
            },
            {
                "name": "fp_non_retryable_create_agent_conflict",
                "expected": {OUTCOME_CONFLICT},
                "payload": self._build_create_payload(
                    tool_payloads=[self._build_flow_payload(label="snap_fp_conflict")],
                    raw_connection=self._build_config_payload(label="cfg_fp_conflict"),
                ),
                "inject": {
                    "create_agent": {
                        "fail_first_n": 1,
                        "error_type": "domain_conflict",
                        "message": "fp_create_agent_conflict",
                    }
                },
            },
            {
                "name": "fp_create_agent_failure_triggers_rollback",
                "expected": {OUTCOME_FAILURE},
                "detail_contains": "Please check server logs for details",
                "payload": self._build_create_payload(
                    tool_payloads=[self._build_flow_payload(label="snap_fp_rollback")],
                    raw_connection=self._build_config_payload(label="cfg_fp_rollback"),
                ),
                "inject": {
                    "create_agent": {
                        "fail_first_n": 3,
                        "error_type": "runtime",
                        "message": "fp_create_agent_final",
                    },
                    "rollback_delete_config": {
                        "fail_first_n": 2,
                        "error_type": "runtime",
                        "message": "fp_rollback_delete_config",
                    },
                },
            },
        ]
        results = await self._run_scenarios(scenarios)
        results.extend(await self._run_update_failpoint_scenarios())
        return results

    async def _run_scenarios(self, scenarios: list[dict[str, Any]]) -> list[ScenarioResult]:
        results: list[ScenarioResult] = []
        for index, scenario in enumerate(scenarios, start=1):
            print(f"\n[{index}/{len(scenarios)}] {scenario['name']}")
            status_code, detail, created = await self._run_create(scenario["payload"], inject=scenario.get("inject"))
            detail_contains = str(scenario.get("detail_contains") or "").strip()
            detail_ok = not detail_contains or detail_contains in detail
            ok = status_code in scenario["expected"] and detail_ok
            dedupe_expected = scenario.get("assert_dedupe_snapshot_count")
            if dedupe_expected is not None:
                created_snapshot_ids = self._extract_create_snapshot_ids(created.provider_result) if created else set()
                created_snapshot_count = len(created_snapshot_ids)
                dedupe_ok = (
                    status_code == OUTCOME_SUCCESS
                    and created is not None
                    and created_snapshot_count == int(dedupe_expected)
                    and self._has_unique_snapshot_ids(created_snapshot_ids)
                )
                ok = ok and dedupe_ok
                if not dedupe_ok:
                    detail = (
                        f"{detail} | dedupe_check failed: expected_snapshot_count={dedupe_expected} "
                        f"got={created_snapshot_count}"
                    )

            if created:
                self.created_deployment_ids.add(created.deployment_id)
                self.created_snapshot_ids.update(self._extract_create_snapshot_ids(created.provider_result))
                self.created_config_ids.update(self._extract_create_app_ids(created.provider_result))

            results.append(
                ScenarioResult(
                    name=scenario["name"],
                    expected_outcomes=set(scenario["expected"]),
                    actual_outcome=status_code,
                    ok=ok,
                    detail=detail[:600],
                )
            )
        return results

    async def _run_create(
        self,
        payload: DeploymentCreate,
        *,
        inject: dict[str, dict[str, Any]] | None = None,
    ) -> tuple[str, str, WxoCreatedDeploymentResult | None]:
        originals: list[tuple[Any, str, Any]] = []
        try:
            if inject:
                self._apply_injections(inject, originals)

            result = await self.service.create(user_id=self.user_id, payload=payload, db=self.db)
        except ResourceConflictError as exc:
            return OUTCOME_CONFLICT, exc.message, None
        except InvalidContentError as exc:
            return OUTCOME_INVALID_CONTENT, exc.message, None
        except (InvalidDeploymentOperationError, InvalidDeploymentTypeError) as exc:
            return OUTCOME_INVALID_OPERATION, exc.message, None
        except DeploymentError as exc:
            return OUTCOME_FAILURE, exc.message, None
        except HTTPException as exc:
            return self._outcome_from_http_exception(exc), str(exc.detail), None
        except Exception as exc:  # noqa: BLE001
            return OUTCOME_FAILURE, str(exc), None
        else:
            create_provider_result = self._parse_create_provider_result_payload(result.provider_result)
            created = WxoCreatedDeploymentResult(
                deployment_id=str(result.id),
                provider_result=create_provider_result,
            )
            return OUTCOME_SUCCESS, "created", created
        finally:
            for target, attr_name, original in originals:
                setattr(target, attr_name, original)

    async def _run_list(self, *, params: DeploymentListParams | None = None) -> tuple[str, str, Any | None]:
        try:
            result = await self.service.list(user_id=self.user_id, db=self.db, params=params)
        except DeploymentNotFoundError as exc:
            return OUTCOME_NOT_FOUND, str(exc), None
        except ResourceConflictError as exc:
            return OUTCOME_CONFLICT, exc.message, None
        except InvalidContentError as exc:
            return OUTCOME_INVALID_CONTENT, exc.message, None
        except (InvalidDeploymentOperationError, InvalidDeploymentTypeError) as exc:
            return OUTCOME_INVALID_OPERATION, exc.message, None
        except DeploymentError as exc:
            return OUTCOME_FAILURE, exc.message, None
        except HTTPException as exc:
            return self._outcome_from_http_exception(exc), str(exc.detail), None
        except Exception as exc:  # noqa: BLE001
            return OUTCOME_FAILURE, str(exc), None
        else:
            return OUTCOME_SUCCESS, "listed", result

    async def _run_get(self, deployment_id: str) -> tuple[str, str, Any | None]:
        try:
            result = await self.service.get(user_id=self.user_id, deployment_id=deployment_id, db=self.db)
        except DeploymentNotFoundError as exc:
            return OUTCOME_NOT_FOUND, str(exc), None
        except ResourceConflictError as exc:
            return OUTCOME_CONFLICT, exc.message, None
        except InvalidContentError as exc:
            return OUTCOME_INVALID_CONTENT, exc.message, None
        except (InvalidDeploymentOperationError, InvalidDeploymentTypeError) as exc:
            return OUTCOME_INVALID_OPERATION, exc.message, None
        except DeploymentError as exc:
            return OUTCOME_FAILURE, exc.message, None
        except HTTPException as exc:
            return self._outcome_from_http_exception(exc), str(exc.detail), None
        except Exception as exc:  # noqa: BLE001
            return OUTCOME_FAILURE, str(exc), None
        else:
            return OUTCOME_SUCCESS, "fetched", result

    async def _run_update(
        self,
        deployment_id: str,
        payload: DeploymentUpdate,
        *,
        inject: dict[str, dict[str, Any]] | None = None,
    ) -> tuple[str, str, Any | None]:
        originals: list[tuple[Any, str, Any]] = []
        try:
            if inject:
                self._apply_injections(inject, originals)
            result = await self.service.update(
                user_id=self.user_id,
                deployment_id=deployment_id,
                payload=payload,
                db=self.db,
            )
        except DeploymentNotFoundError as exc:
            return OUTCOME_NOT_FOUND, str(exc), None
        except ResourceConflictError as exc:
            return OUTCOME_CONFLICT, exc.message, None
        except InvalidContentError as exc:
            return OUTCOME_INVALID_CONTENT, exc.message, None
        except (InvalidDeploymentOperationError, InvalidDeploymentTypeError) as exc:
            return OUTCOME_INVALID_OPERATION, exc.message, None
        except DeploymentError as exc:
            return OUTCOME_FAILURE, exc.message, None
        except HTTPException as exc:
            return self._outcome_from_http_exception(exc), str(exc.detail), None
        except Exception as exc:  # noqa: BLE001
            return OUTCOME_FAILURE, str(exc), None
        else:
            self.created_config_ids.update(self._extract_update_created_app_ids(result))
            return OUTCOME_SUCCESS, "updated", result
        finally:
            for target, attr_name, original in originals:
                setattr(target, attr_name, original)

    async def _run_list_snapshots_with_params(
        self,
        *,
        params: SnapshotListParams | None,
        detail_label: str = "snapshots_listed",
    ) -> tuple[str, str, Any | None]:
        try:
            result = await self.service.list_snapshots(
                user_id=self.user_id,
                params=params,
                db=self.db,
            )
        except DeploymentNotFoundError as exc:
            return OUTCOME_NOT_FOUND, str(exc), None
        except ResourceConflictError as exc:
            return OUTCOME_CONFLICT, exc.message, None
        except InvalidContentError as exc:
            return OUTCOME_INVALID_CONTENT, exc.message, None
        except (InvalidDeploymentOperationError, InvalidDeploymentTypeError) as exc:
            return OUTCOME_INVALID_OPERATION, exc.message, None
        except DeploymentError as exc:
            return OUTCOME_FAILURE, exc.message, None
        except HTTPException as exc:
            return self._outcome_from_http_exception(exc), str(exc.detail), None
        except Exception as exc:  # noqa: BLE001
            return OUTCOME_FAILURE, str(exc), None
        else:
            return OUTCOME_SUCCESS, detail_label, result

    async def _run_list_snapshots(self, deployment_id: str) -> tuple[str, str, Any | None]:
        return await self._run_list_snapshots_with_params(
            params=SnapshotListParams(deployment_ids=[deployment_id]),
            detail_label="snapshots_listed",
        )

    async def _run_list_snapshots_by_ids(self, snapshot_ids: list[str]) -> tuple[str, str, Any | None]:
        return await self._run_list_snapshots_with_params(
            params=SnapshotListParams(snapshot_ids=snapshot_ids),
            detail_label="snapshots_by_ids_listed",
        )

    async def _run_list_snapshots_by_names(self, snapshot_names: list[str]) -> tuple[str, str, Any | None]:
        return await self._run_list_snapshots_with_params(
            params=SnapshotListParams(snapshot_names=snapshot_names),
            detail_label="snapshots_by_names_listed",
        )

    async def _run_list_configs_with_params(
        self,
        *,
        params: ConfigListParams | None,
        detail_label: str = "configs_listed",
    ) -> tuple[str, str, Any | None]:
        try:
            result = await self.service.list_configs(
                user_id=self.user_id,
                params=params,
                db=self.db,
            )
        except DeploymentNotFoundError as exc:
            return OUTCOME_NOT_FOUND, str(exc), None
        except ResourceConflictError as exc:
            return OUTCOME_CONFLICT, exc.message, None
        except InvalidContentError as exc:
            return OUTCOME_INVALID_CONTENT, exc.message, None
        except (InvalidDeploymentOperationError, InvalidDeploymentTypeError) as exc:
            return OUTCOME_INVALID_OPERATION, exc.message, None
        except DeploymentError as exc:
            return OUTCOME_FAILURE, exc.message, None
        except HTTPException as exc:
            return self._outcome_from_http_exception(exc), str(exc.detail), None
        except Exception as exc:  # noqa: BLE001
            return OUTCOME_FAILURE, str(exc), None
        else:
            return OUTCOME_SUCCESS, detail_label, result

    async def _run_list_configs(self, deployment_id: str) -> tuple[str, str, Any | None]:
        return await self._run_list_configs_with_params(
            params=ConfigListParams(deployment_ids=[deployment_id]),
            detail_label="configs_listed",
        )

    async def _run_list_types(self) -> tuple[str, str, Any | None]:
        try:
            result = await self.service.list_types(user_id=self.user_id, db=self.db)
        except DeploymentNotFoundError as exc:
            return OUTCOME_NOT_FOUND, str(exc), None
        except ResourceConflictError as exc:
            return OUTCOME_CONFLICT, exc.message, None
        except InvalidContentError as exc:
            return OUTCOME_INVALID_CONTENT, exc.message, None
        except (InvalidDeploymentOperationError, InvalidDeploymentTypeError, OperationNotSupportedError) as exc:
            return OUTCOME_INVALID_OPERATION, exc.message, None
        except DeploymentError as exc:
            return OUTCOME_FAILURE, exc.message, None
        except HTTPException as exc:
            return self._outcome_from_http_exception(exc), str(exc.detail), None
        except Exception as exc:  # noqa: BLE001
            return OUTCOME_FAILURE, str(exc), None
        else:
            return OUTCOME_SUCCESS, "types_listed", result

    async def _run_list_llms(self) -> tuple[str, str, Any | None]:
        try:
            result = await self.service.list_llms(user_id=self.user_id, db=self.db)
        except DeploymentNotFoundError as exc:
            return OUTCOME_NOT_FOUND, str(exc), None
        except ResourceConflictError as exc:
            return OUTCOME_CONFLICT, exc.message, None
        except InvalidContentError as exc:
            return OUTCOME_INVALID_CONTENT, exc.message, None
        except (InvalidDeploymentOperationError, InvalidDeploymentTypeError, OperationNotSupportedError) as exc:
            return OUTCOME_INVALID_OPERATION, exc.message, None
        except DeploymentError as exc:
            return OUTCOME_FAILURE, exc.message, None
        except HTTPException as exc:
            return self._outcome_from_http_exception(exc), str(exc.detail), None
        except Exception as exc:  # noqa: BLE001
            return OUTCOME_FAILURE, str(exc), None
        else:
            return OUTCOME_SUCCESS, "llms_listed", result

    async def _run_verify_credentials(self, *, base_url: str, api_key: str) -> tuple[str, str, Any | None]:
        try:
            result = await self.service.verify_credentials(
                user_id=self.user_id,
                payload=VerifyCredentials(base_url=base_url, provider_data={"api_key": api_key}),
            )
        except DeploymentNotFoundError as exc:
            return OUTCOME_NOT_FOUND, str(exc), None
        except ResourceConflictError as exc:
            return OUTCOME_CONFLICT, exc.message, None
        except InvalidContentError as exc:
            return OUTCOME_INVALID_CONTENT, exc.message, None
        except (InvalidDeploymentOperationError, InvalidDeploymentTypeError, OperationNotSupportedError) as exc:
            return OUTCOME_INVALID_OPERATION, exc.message, None
        except DeploymentError as exc:
            return OUTCOME_FAILURE, exc.message, None
        except HTTPException as exc:
            return self._outcome_from_http_exception(exc), str(exc.detail), None
        except Exception as exc:  # noqa: BLE001
            return OUTCOME_FAILURE, str(exc), None
        else:
            return OUTCOME_SUCCESS, "credentials_verified", result

    async def _run_update_snapshot(
        self,
        *,
        snapshot_id: str,
        flow_artifact: BaseFlowArtifact[WatsonxFlowArtifactProviderData],
    ) -> tuple[str, str, Any | None]:
        try:
            result = await self.service.update_snapshot(
                user_id=self.user_id,
                db=self.db,
                snapshot_id=snapshot_id,
                flow_artifact=flow_artifact,
            )
        except DeploymentNotFoundError as exc:
            return OUTCOME_NOT_FOUND, str(exc), None
        except ResourceConflictError as exc:
            return OUTCOME_CONFLICT, exc.message, None
        except InvalidContentError as exc:
            return OUTCOME_INVALID_CONTENT, exc.message, None
        except (InvalidDeploymentOperationError, InvalidDeploymentTypeError, OperationNotSupportedError) as exc:
            return OUTCOME_INVALID_OPERATION, exc.message, None
        except DeploymentError as exc:
            return OUTCOME_FAILURE, exc.message, None
        except HTTPException as exc:
            return self._outcome_from_http_exception(exc), str(exc.detail), None
        except Exception as exc:  # noqa: BLE001
            return OUTCOME_FAILURE, str(exc), None
        else:
            return OUTCOME_SUCCESS, "snapshot_updated", result

    async def _run_rollback_create_result(
        self,
        *,
        deployment_id: str,
        provider_result: object,
    ) -> tuple[str, str, Any | None]:
        try:
            result = await self.service.rollback_create_result(
                user_id=self.user_id,
                deployment_id=deployment_id,
                provider_result=provider_result,
                db=self.db,
            )
        except DeploymentNotFoundError as exc:
            return OUTCOME_NOT_FOUND, str(exc), None
        except ResourceConflictError as exc:
            return OUTCOME_CONFLICT, exc.message, None
        except InvalidContentError as exc:
            return OUTCOME_INVALID_CONTENT, exc.message, None
        except (InvalidDeploymentOperationError, InvalidDeploymentTypeError, OperationNotSupportedError) as exc:
            return OUTCOME_INVALID_OPERATION, exc.message, None
        except DeploymentError as exc:
            return OUTCOME_FAILURE, exc.message, None
        except HTTPException as exc:
            return self._outcome_from_http_exception(exc), str(exc.detail), None
        except Exception as exc:  # noqa: BLE001
            return OUTCOME_FAILURE, str(exc), None
        else:
            return OUTCOME_SUCCESS, "rollback_create_result_done", result

    async def _run_redeploy(self, deployment_id: str) -> tuple[str, str, Any | None]:
        try:
            result = await self.service.redeploy(
                user_id=self.user_id,
                deployment_id=deployment_id,
                db=self.db,
            )
        except DeploymentNotFoundError as exc:
            return OUTCOME_NOT_FOUND, str(exc), None
        except ResourceConflictError as exc:
            return OUTCOME_CONFLICT, exc.message, None
        except InvalidContentError as exc:
            return OUTCOME_INVALID_CONTENT, exc.message, None
        except (InvalidDeploymentOperationError, InvalidDeploymentTypeError, OperationNotSupportedError) as exc:
            return OUTCOME_INVALID_OPERATION, exc.message, None
        except DeploymentError as exc:
            return OUTCOME_FAILURE, exc.message, None
        except HTTPException as exc:
            return self._outcome_from_http_exception(exc), str(exc.detail), None
        except Exception as exc:  # noqa: BLE001
            return OUTCOME_FAILURE, str(exc), None
        else:
            return OUTCOME_SUCCESS, "redeployed", result

    async def _run_duplicate(self, deployment_id: str) -> tuple[str, str, Any | None]:
        try:
            result = await self.service.duplicate(
                user_id=self.user_id,
                deployment_id=deployment_id,
                db=self.db,
            )
        except DeploymentNotFoundError as exc:
            return OUTCOME_NOT_FOUND, str(exc), None
        except ResourceConflictError as exc:
            return OUTCOME_CONFLICT, exc.message, None
        except InvalidContentError as exc:
            return OUTCOME_INVALID_CONTENT, exc.message, None
        except (InvalidDeploymentOperationError, InvalidDeploymentTypeError, OperationNotSupportedError) as exc:
            return OUTCOME_INVALID_OPERATION, exc.message, None
        except DeploymentError as exc:
            return OUTCOME_FAILURE, exc.message, None
        except HTTPException as exc:
            return self._outcome_from_http_exception(exc), str(exc.detail), None
        except Exception as exc:  # noqa: BLE001
            return OUTCOME_FAILURE, str(exc), None
        else:
            return OUTCOME_SUCCESS, "duplicated", result

    async def _run_teardown(self) -> tuple[str, str, Any | None]:
        try:
            result = await self.service.teardown()
        except DeploymentNotFoundError as exc:
            return OUTCOME_NOT_FOUND, str(exc), None
        except ResourceConflictError as exc:
            return OUTCOME_CONFLICT, exc.message, None
        except InvalidContentError as exc:
            return OUTCOME_INVALID_CONTENT, exc.message, None
        except (InvalidDeploymentOperationError, InvalidDeploymentTypeError, OperationNotSupportedError) as exc:
            return OUTCOME_INVALID_OPERATION, exc.message, None
        except DeploymentError as exc:
            return OUTCOME_FAILURE, exc.message, None
        except HTTPException as exc:
            return self._outcome_from_http_exception(exc), str(exc.detail), None
        except Exception as exc:  # noqa: BLE001
            return OUTCOME_FAILURE, str(exc), None
        else:
            return OUTCOME_SUCCESS, "teardown_done", result

    async def _run_status(self, deployment_id: str) -> tuple[str, str, Any | None]:
        try:
            result = await self.service.get_status(user_id=self.user_id, deployment_id=deployment_id, db=self.db)
        except DeploymentNotFoundError as exc:
            return OUTCOME_NOT_FOUND, str(exc), None
        except ResourceConflictError as exc:
            return OUTCOME_CONFLICT, exc.message, None
        except InvalidContentError as exc:
            return OUTCOME_INVALID_CONTENT, exc.message, None
        except (InvalidDeploymentOperationError, InvalidDeploymentTypeError) as exc:
            return OUTCOME_INVALID_OPERATION, exc.message, None
        except DeploymentError as exc:
            return OUTCOME_FAILURE, exc.message, None
        except HTTPException as exc:
            return self._outcome_from_http_exception(exc), str(exc.detail), None
        except Exception as exc:  # noqa: BLE001
            return OUTCOME_FAILURE, str(exc), None
        else:
            return OUTCOME_SUCCESS, "status", result

    async def _run_create_execution(
        self,
        deployment_id: str,
        *,
        provider_data: dict[str, Any],
    ) -> tuple[str, str, Any | None]:
        try:
            result = await self.service.create_execution(
                user_id=self.user_id,
                payload=ExecutionCreate(deployment_id=deployment_id, provider_data=provider_data),
                db=self.db,
            )
        except DeploymentNotFoundError as exc:
            return OUTCOME_NOT_FOUND, str(exc), None
        except ResourceConflictError as exc:
            return OUTCOME_CONFLICT, exc.message, None
        except InvalidContentError as exc:
            return OUTCOME_INVALID_CONTENT, exc.message, None
        except (InvalidDeploymentOperationError, InvalidDeploymentTypeError) as exc:
            return OUTCOME_INVALID_OPERATION, exc.message, None
        except DeploymentError as exc:
            return OUTCOME_FAILURE, exc.message, None
        except HTTPException as exc:
            return self._outcome_from_http_exception(exc), str(exc.detail), None
        except Exception as exc:  # noqa: BLE001
            return OUTCOME_FAILURE, str(exc), None
        else:
            return OUTCOME_SUCCESS, "execution_created", result

    async def _run_get_execution(self, execution_id: str) -> tuple[str, str, Any | None]:
        try:
            result = await self.service.get_execution(user_id=self.user_id, execution_id=execution_id, db=self.db)
        except DeploymentNotFoundError as exc:
            return OUTCOME_NOT_FOUND, str(exc), None
        except ResourceConflictError as exc:
            return OUTCOME_CONFLICT, exc.message, None
        except InvalidContentError as exc:
            return OUTCOME_INVALID_CONTENT, exc.message, None
        except (InvalidDeploymentOperationError, InvalidDeploymentTypeError) as exc:
            return OUTCOME_INVALID_OPERATION, exc.message, None
        except DeploymentError as exc:
            return OUTCOME_FAILURE, exc.message, None
        except HTTPException as exc:
            return self._outcome_from_http_exception(exc), str(exc.detail), None
        except Exception as exc:  # noqa: BLE001
            return OUTCOME_FAILURE, str(exc), None
        else:
            return OUTCOME_SUCCESS, "execution_fetched", result

    async def _poll_execution_terminal(self, execution_id: str) -> tuple[str, str, Any | None]:
        """Poll get_execution until a terminal status is reached or max attempts exceeded."""
        result = None
        for attempt in range(EXECUTION_POLL_MAX_ATTEMPTS):
            status_code, detail, result = await self._run_get_execution(execution_id)
            pr = getattr(result, "provider_result", None) or {}
            current_status = pr.get("status") if isinstance(pr, dict) else None
            print(
                f"  [poll {attempt + 1}/{EXECUTION_POLL_MAX_ATTEMPTS}] "
                f"outcome={status_code} status={current_status} "
                f"provider_result={pr}"
            )
            if status_code != OUTCOME_SUCCESS:
                return status_code, f"poll attempt {attempt + 1}: {detail}", result
            if current_status in EXECUTION_TERMINAL_STATUSES:
                return status_code, f"terminal after {attempt + 1} polls: {current_status}", result
            await asyncio.sleep(EXECUTION_POLL_INTERVAL_SECS)
        msg = f"execution did not reach terminal status after {EXECUTION_POLL_MAX_ATTEMPTS} polls"
        return OUTCOME_FAILURE, msg, result

    async def _run_delete(self, deployment_id: str) -> tuple[str, str, Any | None]:
        try:
            result = await self.service.delete(user_id=self.user_id, deployment_id=deployment_id, db=self.db)
        except DeploymentNotFoundError as exc:
            return OUTCOME_NOT_FOUND, str(exc), None
        except ResourceConflictError as exc:
            return OUTCOME_CONFLICT, exc.message, None
        except InvalidContentError as exc:
            return OUTCOME_INVALID_CONTENT, exc.message, None
        except (InvalidDeploymentOperationError, InvalidDeploymentTypeError) as exc:
            return OUTCOME_INVALID_OPERATION, exc.message, None
        except DeploymentError as exc:
            return OUTCOME_FAILURE, exc.message, None
        except HTTPException as exc:
            return self._outcome_from_http_exception(exc), str(exc.detail), None
        except Exception as exc:  # noqa: BLE001
            return OUTCOME_FAILURE, str(exc), None
        else:
            return OUTCOME_SUCCESS, "deleted", result

    def _build_result(
        self,
        *,
        name: str,
        expected: set[str],
        actual_outcome: str,
        detail: str,
        ok: bool,
    ) -> ScenarioResult:
        return ScenarioResult(
            name=name,
            expected_outcomes=expected,
            actual_outcome=actual_outcome,
            ok=ok,
            detail=detail[:600],
        )

    def _outcome_from_http_exception(self, exc: HTTPException) -> str:
        status_code = int(exc.status_code)
        if status_code == HTTP_STATUS_NOT_FOUND:
            return OUTCOME_NOT_FOUND
        if status_code == HTTP_STATUS_CONFLICT:
            return OUTCOME_CONFLICT
        if status_code in {400, 405}:
            return OUTCOME_INVALID_OPERATION
        if status_code in {413, 415, 422}:
            return OUTCOME_INVALID_CONTENT
        return OUTCOME_FAILURE

    async def _run_live_lifecycle_scenarios(self) -> list[ScenarioResult]:
        results: list[ScenarioResult] = []
        print("\n[life/1] live_lifecycle_create_seed")
        status_code, detail, created = await self._run_create(
            self._build_create_payload(
                tool_payloads=[self._build_flow_payload(label="snap_live_lifecycle_seed")],
                raw_connection=self._build_config_payload(label="cfg_live_lifecycle_seed"),
            )
        )
        create_ok = status_code == OUTCOME_SUCCESS and created is not None
        results.append(
            self._build_result(
                name="live_lifecycle_create_seed",
                expected={OUTCOME_SUCCESS},
                actual_outcome=status_code,
                detail=detail,
                ok=create_ok,
            )
        )
        if not create_ok or created is None:
            return results

        deployment_id = created.deployment_id
        self.created_deployment_ids.add(deployment_id)
        self.created_snapshot_ids.update(self._extract_create_snapshot_ids(created.provider_result))
        self.created_config_ids.update(self._extract_create_app_ids(created.provider_result))

        print("[life/2] live_list_contains_seed")
        status_code, detail, list_result = await self._run_list(
            params=DeploymentListParams(deployment_ids=[deployment_id])
        )
        list_contains_seed = bool(
            list_result
            and any(str(deployment.id) == deployment_id for deployment in getattr(list_result, "deployments", []))
        )
        results.append(
            self._build_result(
                name="live_list_contains_seed",
                expected={OUTCOME_SUCCESS},
                actual_outcome=status_code,
                detail=detail,
                ok=status_code == OUTCOME_SUCCESS and list_contains_seed,
            )
        )

        print("[life/3] live_get_seed")
        status_code, detail, get_result = await self._run_get(deployment_id)
        got_seed = bool(get_result and str(get_result.id) == deployment_id)
        results.append(
            self._build_result(
                name="live_get_seed",
                expected={OUTCOME_SUCCESS},
                actual_outcome=status_code,
                detail=detail,
                ok=status_code == OUTCOME_SUCCESS and got_seed,
            )
        )

        updated_name = self._mk_name("dep_agent_updated")
        print("[life/4] live_update_seed_name_description")
        status_code, detail, _ = await self._run_update(
            deployment_id,
            DeploymentUpdate(
                spec=BaseDeploymentDataUpdate(
                    name=updated_name,
                    description="updated by direct adapter e2e",
                )
            ),
        )
        results.append(
            self._build_result(
                name="live_update_seed_name_description",
                expected={OUTCOME_SUCCESS},
                actual_outcome=status_code,
                detail=detail,
                ok=status_code == OUTCOME_SUCCESS,
            )
        )

        print("[life/5] live_get_after_update_reflects_name")
        status_code, detail, get_after_update = await self._run_get(deployment_id)
        updated_name_ok = bool(get_after_update and getattr(get_after_update, "name", None) == updated_name)
        results.append(
            self._build_result(
                name="live_get_after_update_reflects_name",
                expected={OUTCOME_SUCCESS},
                actual_outcome=status_code,
                detail=detail,
                ok=status_code == OUTCOME_SUCCESS and updated_name_ok,
            )
        )

        print("[life/6] live_get_status_connected")
        status_code, detail, status_result = await self._run_status(deployment_id)
        connected_ok = bool(status_result and getattr(status_result, "provider_data", {}).get("status") == "connected")
        results.append(
            self._build_result(
                name="live_get_status_connected",
                expected={OUTCOME_SUCCESS},
                actual_outcome=status_code,
                detail=detail,
                ok=status_code == OUTCOME_SUCCESS and connected_ok,
            )
        )

        print("[life/7] live_create_execution_success")
        status_code, detail, execution_create_result = await self._run_create_execution(
            deployment_id,
            provider_data={"message": {"role": "user", "content": "hi"}},
        )
        has_execution_id = bool(execution_create_result and getattr(execution_create_result, "execution_id", None))
        create_pr = getattr(execution_create_result, "provider_result", None) or {}
        create_pr_ok = (
            has_execution_id
            and isinstance(create_pr, dict)
            and "execution_id" in create_pr
            and "run_id" not in create_pr
            and create_pr.get("status") is not None
        )
        results.append(
            self._build_result(
                name="live_create_execution_success",
                expected={OUTCOME_SUCCESS},
                actual_outcome=status_code,
                detail=detail,
                ok=status_code == OUTCOME_SUCCESS and create_pr_ok,
            )
        )

        execution_id_value = (
            execution_create_result.execution_id
            if execution_create_result and getattr(execution_create_result, "execution_id", None)
            else None
        )
        execution_id = str(execution_id_value) if execution_id_value else None

        print("[life/7b] live_create_execution_input_string")
        status_code, detail, exec_str_result = await self._run_create_execution(
            deployment_id,
            provider_data={"input": "hi again"},
        )
        str_ok = bool(
            status_code == OUTCOME_SUCCESS and exec_str_result and getattr(exec_str_result, "execution_id", None)
        )
        results.append(
            self._build_result(
                name="live_create_execution_input_string",
                expected={OUTCOME_SUCCESS},
                actual_outcome=status_code,
                detail=detail,
                ok=str_ok,
            )
        )

        print("[life/7c] live_create_execution_input_dict_content")
        status_code, detail, exec_dict_result = await self._run_create_execution(
            deployment_id,
            provider_data={"input": {"content": "haha"}},
        )
        dict_ok = bool(
            status_code == OUTCOME_SUCCESS and exec_dict_result and getattr(exec_dict_result, "execution_id", None)
        )
        results.append(
            self._build_result(
                name="live_create_execution_input_dict_content",
                expected={OUTCOME_SUCCESS},
                actual_outcome=status_code,
                detail=detail,
                ok=dict_ok,
            )
        )

        if execution_id:
            print("[life/8] live_get_execution_poll_terminal")
            terminal_result = await self._poll_execution_terminal(execution_id)
            poll_status_code = terminal_result[0]
            poll_detail = terminal_result[1]
            poll_result = terminal_result[2]
            poll_pr = getattr(poll_result, "provider_result", None) or {}
            got_terminal = isinstance(poll_pr, dict) and poll_pr.get("status") in EXECUTION_TERMINAL_STATUSES
            pr_has_execution_id = isinstance(poll_pr, dict) and "execution_id" in poll_pr
            pr_no_run_id = isinstance(poll_pr, dict) and "run_id" not in poll_pr
            results.append(
                self._build_result(
                    name="live_get_execution_poll_terminal",
                    expected={OUTCOME_SUCCESS},
                    actual_outcome=poll_status_code,
                    detail=poll_detail,
                    ok=poll_status_code == OUTCOME_SUCCESS and got_terminal and pr_has_execution_id and pr_no_run_id,
                )
            )

            print("[life/8b] live_get_execution_terminal_fields")
            terminal_fields_ok = (
                got_terminal
                and isinstance(poll_pr, dict)
                and poll_pr.get("agent_id") is not None
                and (
                    poll_pr.get("completed_at") is not None
                    or poll_pr.get("failed_at") is not None
                    or poll_pr.get("cancelled_at") is not None
                )
            )
            results.append(
                self._build_result(
                    name="live_get_execution_terminal_fields",
                    expected={OUTCOME_SUCCESS},
                    actual_outcome=poll_status_code,
                    detail=f"status={poll_pr.get('status')} has_timestamps={terminal_fields_ok}",
                    ok=poll_status_code == OUTCOME_SUCCESS and terminal_fields_ok,
                )
            )

        print("[life/9] live_delete_seed")
        status_code, detail, _ = await self._run_delete(deployment_id)
        delete_ok = status_code == OUTCOME_SUCCESS
        if delete_ok:
            self.created_deployment_ids.discard(deployment_id)
        results.append(
            self._build_result(
                name="live_delete_seed",
                expected={OUTCOME_SUCCESS},
                actual_outcome=status_code,
                detail=detail,
                ok=delete_ok,
            )
        )

        print("[life/10] live_get_after_delete_not_found")
        status_code, detail, _ = await self._run_get(deployment_id)
        results.append(
            self._build_result(
                name="live_get_after_delete_not_found",
                expected={OUTCOME_NOT_FOUND},
                actual_outcome=status_code,
                detail=detail,
                ok=status_code == OUTCOME_NOT_FOUND,
            )
        )

        print("[life/11] live_status_after_delete_not_found_state")
        status_code, detail, _ = await self._run_status(deployment_id)
        results.append(
            self._build_result(
                name="live_status_after_delete_not_found_state",
                expected={OUTCOME_NOT_FOUND},
                actual_outcome=status_code,
                detail=detail,
                ok=status_code == OUTCOME_NOT_FOUND,
            )
        )

        return results

    async def _run_live_list_snapshots_by_ids_scenarios(self) -> list[ScenarioResult]:
        results: list[ScenarioResult] = []
        print("\n[snap-ids/1] creating seed for list_snapshots by-ids mode")
        _deployment_id, _config_id, seed_snapshot_ids, _ = await self._create_update_seed(
            label="snap_ids_seed",
            snapshot_count=2,
        )
        known_ids = sorted(seed_snapshot_ids)
        if len(known_ids) < MIN_MIXED_SNAPSHOT_IDS:
            results.append(
                self._build_result(
                    name="live_list_snapshots_by_ids_seed_insufficient",
                    expected={OUTCOME_SUCCESS},
                    actual_outcome=OUTCOME_FAILURE,
                    detail=f"need >= {MIN_MIXED_SNAPSHOT_IDS} snapshot ids, got {len(known_ids)}",
                    ok=False,
                )
            )
            return results

        print("[snap-ids/2] live_list_snapshots_by_ids_returns_known")
        status_code, detail, snap_result = await self._run_list_snapshots_by_ids(known_ids)
        returned_ids = self._extract_snapshot_ids(snap_result)
        ids_match = set(known_ids) == returned_ids
        results.append(
            self._build_result(
                name="live_list_snapshots_by_ids_returns_known",
                expected={OUTCOME_SUCCESS},
                actual_outcome=status_code,
                detail=f"{detail} | returned={sorted(returned_ids)} expected={known_ids}",
                ok=status_code == OUTCOME_SUCCESS and ids_match,
            )
        )

        print("[snap-ids/3] live_list_snapshots_by_ids_filters_unknown")
        bogus_id = str(uuid4())
        mixed_ids = [known_ids[0], bogus_id]
        status_code, detail, snap_result = await self._run_list_snapshots_by_ids(mixed_ids)
        returned_ids = self._extract_snapshot_ids(snap_result)
        has_known = known_ids[0] in returned_ids
        no_bogus = bogus_id not in returned_ids
        results.append(
            self._build_result(
                name="live_list_snapshots_by_ids_filters_unknown",
                expected={OUTCOME_SUCCESS},
                actual_outcome=status_code,
                detail=f"{detail} | returned={sorted(returned_ids)} has_known={has_known} no_bogus={no_bogus}",
                ok=status_code == OUTCOME_SUCCESS and has_known and no_bogus,
            )
        )

        # Empty snapshot_ids currently normalizes to tenant-scoped listing.
        # We still expect a successful response and ensure known seed IDs
        # are visible in the returned set.
        print("[snap-ids/4] live_list_snapshots_by_ids_empty_input")
        status_code, detail, snap_result = await self._run_list_snapshots_by_ids([])
        returned_ids = self._extract_snapshot_ids(snap_result)
        has_known_subset = set(known_ids).issubset(returned_ids)
        results.append(
            self._build_result(
                name="live_list_snapshots_by_ids_empty_input",
                expected={OUTCOME_SUCCESS},
                actual_outcome=status_code,
                detail=(f"{detail} | returned_count={len(returned_ids)} has_known_subset={has_known_subset}"),
                ok=status_code == OUTCOME_SUCCESS and has_known_subset,
            )
        )

        return results

    async def _run_live_listing_mode_scenarios(self) -> list[ScenarioResult]:
        results: list[ScenarioResult] = []
        print("\n[list/1] creating seed for list mode checks")
        deployment_id, config_id, seed_snapshot_ids, _ = await self._create_update_seed(
            label="list_modes_seed",
            snapshot_count=1,
        )
        seed_snapshot_id = next(iter(seed_snapshot_ids), "")
        if not seed_snapshot_id:
            results.append(
                self._build_result(
                    name="live_list_modes_seed_missing_snapshot",
                    expected={OUTCOME_SUCCESS},
                    actual_outcome=OUTCOME_FAILURE,
                    detail="seed snapshot id missing",
                    ok=False,
                )
            )
            return results

        print("[list/2] live_list_snapshots_tenant_scope")
        status_code, detail, tenant_snapshots = await self._run_list_snapshots_with_params(
            params=None,
            detail_label="snapshots_tenant_listed",
        )
        tenant_snapshot_ids = self._extract_snapshot_ids(tenant_snapshots)
        tenant_has_seed = seed_snapshot_id in tenant_snapshot_ids
        results.append(
            self._build_result(
                name="live_list_snapshots_tenant_scope",
                expected={OUTCOME_SUCCESS},
                actual_outcome=status_code,
                detail=(
                    f"{detail} | seed_snapshot_id={seed_snapshot_id} "
                    f"tenant_has_seed={tenant_has_seed} total={len(tenant_snapshot_ids)}"
                ),
                ok=status_code == OUTCOME_SUCCESS and tenant_has_seed,
            )
        )

        print("[list/3] live_list_configs_tenant_scope")
        status_code, detail, tenant_configs = await self._run_list_configs_with_params(
            params=None,
            detail_label="configs_tenant_listed",
        )
        tenant_config_ids = self._extract_config_ids(tenant_configs)
        tenant_has_config = bool(config_id and config_id in tenant_config_ids)
        results.append(
            self._build_result(
                name="live_list_configs_tenant_scope",
                expected={OUTCOME_SUCCESS},
                actual_outcome=status_code,
                detail=(
                    f"{detail} | seed_config_id={config_id} "
                    f"tenant_has_config={tenant_has_config} total={len(tenant_config_ids)}"
                ),
                ok=status_code == OUTCOME_SUCCESS and bool(config_id) and tenant_has_config,
            )
        )

        deployment_list_status, deployment_list_detail, deployment_snapshot_list = await self._run_list_snapshots(
            deployment_id
        )
        deployment_snapshots = getattr(deployment_snapshot_list, "snapshots", []) if deployment_snapshot_list else []
        seed_snapshot_name = ""
        for snapshot in deployment_snapshots:
            snapshot_id = str(getattr(snapshot, "id", "")).strip()
            snapshot_name = str(getattr(snapshot, "name", "")).strip()
            if snapshot_id == seed_snapshot_id and snapshot_name:
                seed_snapshot_name = snapshot_name
                break
        if not seed_snapshot_name:
            results.append(
                self._build_result(
                    name="live_list_snapshots_by_names_seed_missing_name",
                    expected={OUTCOME_SUCCESS},
                    actual_outcome=deployment_list_status,
                    detail=(
                        f"{deployment_list_detail} | seed_snapshot_id={seed_snapshot_id} "
                        "is missing from deployment-scoped snapshot names"
                    ),
                    ok=False,
                )
            )
            return results

        print("[list/4] live_list_snapshots_by_names_returns_known")
        status_code, detail, by_name_result = await self._run_list_snapshots_by_names([seed_snapshot_name])
        by_name_ids = self._extract_snapshot_ids(by_name_result)
        by_name_has_seed = seed_snapshot_id in by_name_ids
        results.append(
            self._build_result(
                name="live_list_snapshots_by_names_returns_known",
                expected={OUTCOME_SUCCESS},
                actual_outcome=status_code,
                detail=(
                    f"{detail} | seed_snapshot_name={seed_snapshot_name} "
                    f"seed_snapshot_id={seed_snapshot_id} by_name_has_seed={by_name_has_seed}"
                ),
                ok=status_code == OUTCOME_SUCCESS and by_name_has_seed,
            )
        )

        print("[list/5] live_list_snapshots_by_names_ignored_with_deployment_scope")
        status_code, detail, mixed_filter_result = await self._run_list_snapshots_with_params(
            params=SnapshotListParams(
                deployment_ids=[deployment_id],
                snapshot_names=[self._mk_name("snap_name_ignored")],
            ),
            detail_label="snapshots_mixed_filter_listed",
        )
        mixed_filter_ids = self._extract_snapshot_ids(mixed_filter_result)
        mixed_kept_seed = seed_snapshot_id in mixed_filter_ids
        results.append(
            self._build_result(
                name="live_list_snapshots_by_names_ignored_with_deployment_scope",
                expected={OUTCOME_SUCCESS},
                actual_outcome=status_code,
                detail=(
                    f"{detail} | seed_snapshot_id={seed_snapshot_id} mixed_kept_seed={mixed_kept_seed} "
                    f"returned={sorted(mixed_filter_ids)}"
                ),
                ok=status_code == OUTCOME_SUCCESS and mixed_kept_seed,
            )
        )

        return results

    async def _run_live_service_surface_scenarios(self) -> list[ScenarioResult]:
        results: list[ScenarioResult] = []

        print("\n[surface/1] live_list_types_supports_agent")
        status_code, detail, types_result = await self._run_list_types()
        deployment_types = {
            (dtype.value if hasattr(dtype, "value") else str(dtype)).strip()
            for dtype in getattr(types_result, "deployment_types", [])
        }
        has_agent_type = DeploymentType.AGENT.value in deployment_types
        results.append(
            self._build_result(
                name="live_list_types_supports_agent",
                expected={OUTCOME_SUCCESS},
                actual_outcome=status_code,
                detail=f"{detail} | deployment_types={sorted(deployment_types)}",
                ok=status_code == OUTCOME_SUCCESS and has_agent_type,
            )
        )

        print("[surface/2] live_list_llms_returns_models")
        status_code, detail, llms_result = await self._run_list_llms()
        llm_provider_result = getattr(llms_result, "provider_result", {}) if llms_result else {}
        models = llm_provider_result.get("models", []) if isinstance(llm_provider_result, dict) else []
        results.append(
            self._build_result(
                name="live_list_llms_returns_models",
                expected={OUTCOME_SUCCESS},
                actual_outcome=status_code,
                detail=f"{detail} | model_count={len(models)}",
                ok=status_code == OUTCOME_SUCCESS and len(models) > 0,
            )
        )

        print("[surface/3] live_verify_credentials_success")
        status_code, detail, _ = await self._run_verify_credentials(
            base_url=self.provider_backend_url,
            api_key=self.provider_api_key,
        )
        results.append(
            self._build_result(
                name="live_verify_credentials_success",
                expected={OUTCOME_SUCCESS},
                actual_outcome=status_code,
                detail=detail,
                ok=status_code == OUTCOME_SUCCESS,
            )
        )

        print("[surface/4] creating seed for update_snapshot + unsupported operations")
        deployment_id, _config_id, surface_snapshot_ids, _ = await self._create_update_seed(
            label="surface_seed",
            snapshot_count=1,
        )
        surface_snapshot_id = next(iter(surface_snapshot_ids), "")
        if not surface_snapshot_id:
            results.append(
                self._build_result(
                    name="live_update_snapshot_seed_missing_snapshot",
                    expected={OUTCOME_SUCCESS},
                    actual_outcome=OUTCOME_FAILURE,
                    detail="surface seed has no snapshot ids",
                    ok=False,
                )
            )
            return results

        print("[surface/5] live_update_snapshot_success")
        status_code, detail, update_snapshot_result = await self._run_update_snapshot(
            snapshot_id=surface_snapshot_id,
            flow_artifact=self._build_flow_payload(label="surface_update_snapshot_flow"),
        )
        updated_snapshot_id = (
            str(getattr(update_snapshot_result, "snapshot_id", "")).strip() if update_snapshot_result else ""
        )
        results.append(
            self._build_result(
                name="live_update_snapshot_success",
                expected={OUTCOME_SUCCESS},
                actual_outcome=status_code,
                detail=f"{detail} | expected_snapshot_id={surface_snapshot_id} got={updated_snapshot_id}",
                ok=status_code == OUTCOME_SUCCESS and updated_snapshot_id == surface_snapshot_id,
            )
        )

        print("[surface/6] live_rollback_create_result_cleans_up_created")
        rollback_create_status, rollback_create_detail, rollback_created = await self._run_create(
            self._build_create_payload(
                tool_payloads=[self._build_flow_payload(label="surface_rb_seed_snap")],
                raw_connection=self._build_config_payload(label="surface_rb_seed_cfg"),
            )
        )
        if rollback_create_status != OUTCOME_SUCCESS or rollback_created is None:
            results.append(
                self._build_result(
                    name="live_rollback_create_result_cleans_up_created",
                    expected={OUTCOME_SUCCESS},
                    actual_outcome=rollback_create_status,
                    detail=f"seed_create={rollback_create_status}:{rollback_create_detail}",
                    ok=False,
                )
            )
        else:
            rollback_deployment_id = rollback_created.deployment_id
            rollback_snapshot_ids = self._extract_create_snapshot_ids(rollback_created.provider_result)
            rollback_app_ids = self._extract_create_app_ids(rollback_created.provider_result)
            self.created_deployment_ids.add(rollback_deployment_id)
            self.created_snapshot_ids.update(rollback_snapshot_ids)
            self.created_config_ids.update(rollback_app_ids)
            rollback_status, rollback_detail, _ = await self._run_rollback_create_result(
                deployment_id=rollback_deployment_id,
                provider_result=rollback_created.provider_result,
            )
            post_status, post_detail, _ = await self._run_get(rollback_deployment_id)
            cleaned_up = rollback_status == OUTCOME_SUCCESS and post_status == OUTCOME_NOT_FOUND
            if cleaned_up:
                self.created_deployment_ids.discard(rollback_deployment_id)
                self.created_snapshot_ids.difference_update(rollback_snapshot_ids)
                self.created_config_ids.difference_update(rollback_app_ids)
            results.append(
                self._build_result(
                    name="live_rollback_create_result_cleans_up_created",
                    expected={OUTCOME_SUCCESS},
                    actual_outcome=rollback_status,
                    detail=(f"rollback={rollback_status}:{rollback_detail} post_get={post_status}:{post_detail}"),
                    ok=cleaned_up,
                )
            )

        print("[surface/7] live_redeploy_not_supported")
        status_code, detail, _ = await self._run_redeploy(deployment_id)
        redeploy_not_supported = "not supported" in detail.lower()
        results.append(
            self._build_result(
                name="live_redeploy_not_supported",
                expected={OUTCOME_INVALID_OPERATION},
                actual_outcome=status_code,
                detail=detail,
                ok=status_code == OUTCOME_INVALID_OPERATION and redeploy_not_supported,
            )
        )

        print("[surface/8] live_duplicate_not_supported")
        status_code, detail, _ = await self._run_duplicate(deployment_id)
        duplicate_not_supported = "not supported" in detail.lower()
        results.append(
            self._build_result(
                name="live_duplicate_not_supported",
                expected={OUTCOME_INVALID_OPERATION},
                actual_outcome=status_code,
                detail=detail,
                ok=status_code == OUTCOME_INVALID_OPERATION and duplicate_not_supported,
            )
        )

        print("[surface/9] live_teardown_noop")
        status_code, detail, _ = await self._run_teardown()
        results.append(
            self._build_result(
                name="live_teardown_noop",
                expected={OUTCOME_SUCCESS},
                actual_outcome=status_code,
                detail=detail,
                ok=status_code == OUTCOME_SUCCESS,
            )
        )

        return results

    async def _run_live_negative_scenarios(self) -> list[ScenarioResult]:
        results: list[ScenarioResult] = []
        print("\n[neg/1] live_negative_create_seed")
        status_code, detail, created = await self._run_create(
            self._build_create_payload(
                tool_payloads=[self._build_flow_payload(label="snap_live_negative_seed")],
                raw_connection=self._build_config_payload(label="cfg_live_negative_seed"),
            )
        )
        create_ok = status_code == OUTCOME_SUCCESS and created is not None
        results.append(
            self._build_result(
                name="live_negative_create_seed",
                expected={OUTCOME_SUCCESS},
                actual_outcome=status_code,
                detail=detail,
                ok=create_ok,
            )
        )
        if not create_ok or created is None:
            return results

        deployment_id = created.deployment_id
        self.created_deployment_ids.add(deployment_id)
        self.created_snapshot_ids.update(self._extract_create_snapshot_ids(created.provider_result))
        self.created_config_ids.update(self._extract_create_app_ids(created.provider_result))

        print("[neg/2] live_create_execution_rejects_empty_input")
        status_code, detail, _ = await self._run_create_execution(
            deployment_id,
            provider_data={"input": "   "},
        )
        results.append(
            self._build_result(
                name="live_create_execution_rejects_empty_input",
                expected={OUTCOME_INVALID_CONTENT},
                actual_outcome=status_code,
                detail=detail,
                ok=status_code == OUTCOME_INVALID_CONTENT,
            )
        )

        print("[neg/2b] live_create_execution_missing_deployment")
        status_code, detail, _ = await self._run_create_execution(
            str(uuid4()),
            provider_data={"input": "uh oh"},
        )
        results.append(
            self._build_result(
                name="live_create_execution_missing_deployment",
                expected={OUTCOME_NOT_FOUND},
                actual_outcome=status_code,
                detail=detail,
                ok=status_code == OUTCOME_NOT_FOUND,
            )
        )

        print("[neg/3] live_delete_missing_not_found")
        status_code, detail, _ = await self._run_delete(str(uuid4()))
        results.append(
            self._build_result(
                name="live_delete_missing_not_found",
                expected={OUTCOME_NOT_FOUND},
                actual_outcome=status_code,
                detail=detail,
                ok=status_code == OUTCOME_NOT_FOUND,
            )
        )

        print("[neg/4] live_negative_delete_seed")
        status_code, detail, _ = await self._run_delete(deployment_id)
        delete_ok = status_code == OUTCOME_SUCCESS
        if delete_ok:
            self.created_deployment_ids.discard(deployment_id)
        results.append(
            self._build_result(
                name="live_negative_delete_seed",
                expected={OUTCOME_SUCCESS},
                actual_outcome=status_code,
                detail=detail,
                ok=delete_ok,
            )
        )

        return results

    def _extract_snapshot_ids(self, snapshot_result: Any) -> set[str]:
        snapshots = getattr(snapshot_result, "snapshots", []) if snapshot_result else []
        return {str(snapshot.id) for snapshot in snapshots if snapshot and snapshot.id}

    def _parse_create_provider_result_payload(self, provider_result: Any) -> WatsonxDeploymentCreateResultData:
        create_result_slot = self.service.payload_schemas.deployment_create_result
        if create_result_slot is None:
            return WatsonxDeploymentCreateResultData()
        try:
            parsed_result = create_result_slot.parse(provider_result)
        except Exception:  # noqa: BLE001
            return WatsonxDeploymentCreateResultData()
        if isinstance(parsed_result, WatsonxDeploymentCreateResultData):
            return parsed_result
        return WatsonxDeploymentCreateResultData.model_validate(parsed_result)

    def _extract_create_app_ids(self, create_provider_result: WatsonxDeploymentCreateResultData) -> list[str]:
        app_ids = [str(app_id).strip() for app_id in create_provider_result.app_ids if str(app_id).strip()]
        return list(dict.fromkeys(app_ids))

    def _extract_create_snapshot_ids(self, create_provider_result: WatsonxDeploymentCreateResultData) -> set[str]:
        bindings = create_provider_result.tools_with_refs
        snapshot_ids: set[str] = set()
        for binding in bindings:
            tool_id = str(binding.tool_id or "").strip()
            if tool_id:
                snapshot_ids.add(tool_id)
        return snapshot_ids

    def _extract_create_tool_ref_map(
        self,
        create_provider_result: WatsonxDeploymentCreateResultData,
    ) -> dict[str, str]:
        """Return tool_id → source_ref mapping from create result and cache it."""
        ref_map: dict[str, str] = {}
        for binding in create_provider_result.tools_with_refs:
            tool_id = str(binding.tool_id or "").strip()
            source_ref = str(binding.source_ref or "").strip()
            if tool_id and source_ref:
                ref_map[tool_id] = source_ref
        self.tool_source_refs.update(ref_map)
        return ref_map

    def _make_tool_ref(self, tool_id: str) -> dict[str, str]:
        """Build a WatsonxToolRefBinding dict using the cached source_ref mapping."""
        source_ref = self.tool_source_refs.get(tool_id)
        if source_ref is None:
            msg = f"tool_id={tool_id!r} has no cached source_ref; was it created via _create_update_seed?"
            raise ValueError(msg)
        return {"source_ref": source_ref, "tool_id": tool_id}

    def _make_tool_id_with_ref(self, tool_id: str) -> dict[str, Any]:
        """Build a bind-operation tool selector dict using the cached source_ref mapping."""
        return {"tool_id_with_ref": self._make_tool_ref(tool_id)}

    def _extract_update_snapshot_ids(self, update_result: Any) -> set[str]:
        if update_result is None:
            return set()
        provider_result = getattr(update_result, "provider_result", None)
        if isinstance(provider_result, dict):
            snapshot_ids = provider_result.get("created_snapshot_ids", [])
        else:
            snapshot_ids = getattr(provider_result, "created_snapshot_ids", []) if provider_result else []
        return {str(snapshot_id) for snapshot_id in snapshot_ids if str(snapshot_id).strip()}

    def _extract_update_added_snapshot_ids(self, update_result: Any) -> set[str]:
        if update_result is None:
            return set()
        provider_result = getattr(update_result, "provider_result", None)
        if isinstance(provider_result, dict):
            snapshot_ids = provider_result.get("added_snapshot_ids", [])
        else:
            snapshot_ids = getattr(provider_result, "added_snapshot_ids", []) if provider_result else []
        return {str(snapshot_id) for snapshot_id in snapshot_ids if str(snapshot_id).strip()}

    def _extract_update_created_app_ids(self, update_result: Any) -> set[str]:
        if update_result is None:
            return set()
        provider_result = getattr(update_result, "provider_result", None)
        if isinstance(provider_result, dict):
            app_ids = provider_result.get("created_app_ids", [])
        else:
            app_ids = getattr(provider_result, "created_app_ids", []) if provider_result else []
        return {str(app_id).strip() for app_id in app_ids if str(app_id).strip()}

    async def _create_update_seed(
        self,
        *,
        label: str,
        snapshot_count: int = 2,
    ) -> tuple[str, str | None, set[str], str]:
        snapshots = [self._build_flow_payload(label=f"{label}_snap_{idx}") for idx in range(snapshot_count)]
        status_code, detail, created = await self._run_create(
            self._build_create_payload(
                tool_payloads=snapshots,
                raw_connection=self._build_config_payload(label=f"{label}_cfg"),
            )
        )
        if status_code != OUTCOME_SUCCESS or not created:
            msg = f"Failed to create seed deployment '{label}': status={status_code}, detail={detail}"
            raise RuntimeError(msg)
        deployment_id = created.deployment_id
        self.created_deployment_ids.add(deployment_id)
        created_app_ids = self._extract_create_app_ids(created.provider_result)
        created_snapshot_ids = self._extract_create_snapshot_ids(created.provider_result)
        self._extract_create_tool_ref_map(created.provider_result)
        self.created_snapshot_ids.update(created_snapshot_ids)
        self.created_config_ids.update(created_app_ids)
        config_id = created_app_ids[0] if created_app_ids else None
        return deployment_id, config_id, set(created_snapshot_ids), detail

    async def _run_live_update_matrix_scenarios(self) -> list[ScenarioResult]:
        results: list[ScenarioResult] = []
        print("\n[upd] building update matrix seed resources")
        (
            primary_deployment_id,
            _primary_config_id,
            primary_snapshot_ids,
            _,
        ) = await self._create_update_seed(label="upd_primary", snapshot_count=2)
        donor_deployment_id, donor_config_id, donor_snapshot_ids, _ = await self._create_update_seed(
            label="upd_donor",
            snapshot_count=1,
        )
        mixed_donor_deployment_id, _mixed_donor_cfg, mixed_donor_snapshot_ids, _ = await self._create_update_seed(
            label="upd_mixed_donor",
            snapshot_count=1,
        )

        donor_snapshot_id = next(iter(donor_snapshot_ids), "")
        mixed_donor_snapshot_id = next(iter(mixed_donor_snapshot_ids), "")
        removable_snapshot_id = next(iter(primary_snapshot_ids), "")
        retained_snapshot_ids = set(primary_snapshot_ids)
        retained_snapshot_ids.discard(removable_snapshot_id)
        if not donor_config_id:
            results.append(
                self._build_result(
                    name="upd_seed_missing_donor_config",
                    expected={OUTCOME_SUCCESS},
                    actual_outcome=OUTCOME_FAILURE,
                    detail="donor deployment config id is missing",
                    ok=False,
                )
            )
            return results

        print("[upd/1] upd_spec_only_name_desc")
        updated_name = self._mk_name("dep_upd_spec_only")
        status_code, detail, update_result = await self._run_update(
            primary_deployment_id,
            DeploymentUpdate(
                spec=BaseDeploymentDataUpdate(
                    name=updated_name,
                    description="updated by update matrix spec-only",
                )
            ),
        )
        get_status, _get_detail, get_after_update = await self._run_get(primary_deployment_id)
        spec_ok = bool(get_after_update and getattr(get_after_update, "name", None) == updated_name)
        spec_snapshot_ids_ok = bool(update_result and not getattr(update_result, "snapshot_ids", []))
        results.append(
            self._build_result(
                name="upd_spec_only_name_desc",
                expected={OUTCOME_SUCCESS},
                actual_outcome=status_code,
                detail=detail,
                ok=(
                    status_code == OUTCOME_SUCCESS
                    and get_status == OUTCOME_SUCCESS
                    and spec_ok
                    and spec_snapshot_ids_ok
                ),
            )
        )

        print("[upd/2] upd_snapshot_remove_only_no_config")
        status_code, detail, remove_result = await self._run_update(
            primary_deployment_id,
            DeploymentUpdate(
                provider_data={
                    "tools": {},
                    "operations": [{"op": "remove_tool", "tool": self._make_tool_ref(removable_snapshot_id)}],
                }
            ),
        )
        list_status, _list_detail, list_after_remove = await self._run_list_snapshots(primary_deployment_id)
        attached_after_remove = self._extract_snapshot_ids(list_after_remove)
        remove_snapshot_ids_ok = bool(remove_result and not getattr(remove_result, "snapshot_ids", []))
        results.append(
            self._build_result(
                name="upd_snapshot_remove_only_no_config",
                expected={OUTCOME_SUCCESS},
                actual_outcome=status_code,
                detail=detail,
                ok=(
                    status_code == OUTCOME_SUCCESS
                    and list_status == OUTCOME_SUCCESS
                    and removable_snapshot_id not in attached_after_remove
                    and retained_snapshot_ids.issubset(attached_after_remove)
                    and remove_snapshot_ids_ok
                ),
            )
        )

        print("[upd/3] upd_config_only_existing_tools_with_config_id")
        retained_snapshot_ids_sorted = sorted(retained_snapshot_ids)
        status_code, detail, config_only_result = await self._run_update(
            primary_deployment_id,
            DeploymentUpdate(
                provider_data={
                    "tools": {},
                    "connections": {},
                    "operations": [
                        {
                            "op": "bind",
                            "tool": self._make_tool_id_with_ref(tool_id),
                            "app_ids": [str(donor_config_id)],
                        }
                        for tool_id in retained_snapshot_ids_sorted
                    ],
                }
            ),
        )
        list_status, _list_detail, list_after_config_only = await self._run_list_snapshots(primary_deployment_id)
        attached_after_config_only = self._extract_snapshot_ids(list_after_config_only)
        config_only_created_snapshot_ids = self._extract_update_snapshot_ids(config_only_result)
        config_only_created_ok = len(config_only_created_snapshot_ids) == 0
        results.append(
            self._build_result(
                name="upd_config_only_existing_tools_with_config_id",
                expected={OUTCOME_SUCCESS},
                actual_outcome=status_code,
                detail=detail,
                ok=(
                    status_code == OUTCOME_SUCCESS
                    and list_status == OUTCOME_SUCCESS
                    and retained_snapshot_ids.issubset(attached_after_config_only)
                    and config_only_created_ok
                ),
            )
        )

        print("[upd/4] upd_snapshot_add_ids_with_config_id")
        status_code, detail, add_id_result = await self._run_update(
            primary_deployment_id,
            DeploymentUpdate(
                provider_data={
                    "tools": {},
                    "connections": {},
                    "operations": [
                        {
                            "op": "bind",
                            "tool": self._make_tool_id_with_ref(donor_snapshot_id),
                            "app_ids": [str(donor_config_id)],
                        }
                    ],
                }
            ),
        )
        list_status, _list_detail, list_after_add_id = await self._run_list_snapshots(primary_deployment_id)
        attached_after_add_id = self._extract_snapshot_ids(list_after_add_id)
        add_id_snapshot_ids = self._extract_update_added_snapshot_ids(add_id_result)
        results.append(
            self._build_result(
                name="upd_snapshot_add_ids_with_config_id",
                expected={OUTCOME_SUCCESS},
                actual_outcome=status_code,
                detail=detail,
                ok=(
                    status_code == OUTCOME_SUCCESS
                    and list_status == OUTCOME_SUCCESS
                    and donor_snapshot_id in attached_after_add_id
                    and donor_snapshot_id in add_id_snapshot_ids
                ),
            )
        )

        print("[upd/5] upd_snapshot_add_raw_with_config_id")
        raw_payload = self._build_flow_payload(label="upd_add_raw")
        status_code, detail, add_raw_result = await self._run_update(
            primary_deployment_id,
            DeploymentUpdate(
                provider_data={
                    "tools": {"raw_payloads": [raw_payload.model_dump(mode="json")]},
                    "connections": {},
                    "operations": [
                        {
                            "op": "bind",
                            "tool": {"name_of_raw": raw_payload.name},
                            "app_ids": [str(donor_config_id)],
                        }
                    ],
                }
            ),
        )
        add_raw_snapshot_ids = self._extract_update_snapshot_ids(add_raw_result)
        self.created_snapshot_ids.update(add_raw_snapshot_ids)
        list_status, _list_detail, list_after_add_raw = await self._run_list_snapshots(primary_deployment_id)
        attached_after_add_raw = self._extract_snapshot_ids(list_after_add_raw)
        add_raw_created_ids = add_raw_snapshot_ids
        results.append(
            self._build_result(
                name="upd_snapshot_add_raw_with_config_id",
                expected={OUTCOME_SUCCESS},
                actual_outcome=status_code,
                detail=detail,
                ok=(
                    status_code == OUTCOME_SUCCESS
                    and list_status == OUTCOME_SUCCESS
                    and bool(add_raw_created_ids)
                    and add_raw_created_ids.issubset(attached_after_add_raw)
                ),
            )
        )

        print("[upd/6] upd_mixed_add_remove_raw_with_config")
        mixed_raw_payload = self._build_flow_payload(label="upd_mixed_raw")
        mixed_remove_id = next(iter(retained_snapshot_ids), "")
        status_code, detail, mixed_result = await self._run_update(
            primary_deployment_id,
            DeploymentUpdate(
                provider_data={
                    "tools": {
                        "raw_payloads": [mixed_raw_payload.model_dump(mode="json")],
                    },
                    "connections": {},
                    "operations": [
                        {
                            "op": "bind",
                            "tool": self._make_tool_id_with_ref(mixed_donor_snapshot_id),
                            "app_ids": [str(donor_config_id)],
                        },
                        {"op": "remove_tool", "tool": self._make_tool_ref(mixed_remove_id)},
                        {
                            "op": "bind",
                            "tool": {"name_of_raw": mixed_raw_payload.name},
                            "app_ids": [str(donor_config_id)],
                        },
                    ],
                }
            ),
        )
        mixed_snapshot_ids = self._extract_update_added_snapshot_ids(mixed_result)
        self.created_snapshot_ids.update(mixed_snapshot_ids)
        list_status, _list_detail, list_after_mixed = await self._run_list_snapshots(primary_deployment_id)
        attached_after_mixed = self._extract_snapshot_ids(list_after_mixed)
        results.append(
            self._build_result(
                name="upd_mixed_add_remove_raw_with_config",
                expected={OUTCOME_SUCCESS},
                actual_outcome=status_code,
                detail=detail,
                ok=(
                    status_code == OUTCOME_SUCCESS
                    and list_status == OUTCOME_SUCCESS
                    and mixed_remove_id not in attached_after_mixed
                    and mixed_donor_snapshot_id in attached_after_mixed
                    and mixed_donor_snapshot_id in mixed_snapshot_ids
                    and len(mixed_snapshot_ids) >= MIN_MIXED_SNAPSHOT_IDS
                ),
            )
        )

        print("[upd/7] upd_reject_bind_with_undeclared_app_id")
        status_code, detail, _ = await self._run_update(
            primary_deployment_id,
            DeploymentUpdate(
                provider_data={
                    "tools": {},
                    "operations": [
                        {
                            "op": "bind",
                            "tool": self._make_tool_id_with_ref(donor_snapshot_id),
                            "app_ids": ["undeclared_app_for_bind"],
                        }
                    ],
                }
            ),
        )
        results.append(
            self._build_result(
                name="upd_reject_bind_with_undeclared_app_id",
                expected={OUTCOME_INVALID_CONTENT},
                actual_outcome=status_code,
                detail=detail,
                ok=status_code == OUTCOME_INVALID_CONTENT,
            )
        )

        print("[upd/8] upd_reject_raw_bind_with_undeclared_app_id")
        missing_cfg_raw_payload = self._build_flow_payload(label="upd_no_cfg_raw")
        status_code, detail, _ = await self._run_update(
            primary_deployment_id,
            DeploymentUpdate(
                provider_data={
                    "tools": {"raw_payloads": [missing_cfg_raw_payload.model_dump(mode="json")]},
                    "operations": [
                        {
                            "op": "bind",
                            "tool": {"name_of_raw": missing_cfg_raw_payload.name},
                            "app_ids": ["undeclared_app_for_raw_bind"],
                        }
                    ],
                }
            ),
        )
        results.append(
            self._build_result(
                name="upd_reject_raw_bind_with_undeclared_app_id",
                expected={OUTCOME_INVALID_CONTENT},
                actual_outcome=status_code,
                detail=detail,
                ok=status_code == OUTCOME_INVALID_CONTENT,
            )
        )

        print("[upd/9] upd_reject_unbind_with_undeclared_app_id")
        status_code, detail, _ = await self._run_update(
            primary_deployment_id,
            DeploymentUpdate(
                provider_data={
                    "tools": {},
                    "operations": [
                        {
                            "op": "unbind",
                            "tool": self._make_tool_ref(donor_snapshot_id),
                            "app_ids": ["undeclared_app_for_unbind"],
                        }
                    ],
                }
            ),
        )
        results.append(
            self._build_result(
                name="upd_reject_unbind_with_undeclared_app_id",
                expected={OUTCOME_INVALID_CONTENT},
                actual_outcome=status_code,
                detail=detail,
                ok=status_code == OUTCOME_INVALID_CONTENT,
            )
        )

        print("[upd/10] upd_reject_unbind_unknown_tool_id")
        status_code, detail, _ = await self._run_update(
            primary_deployment_id,
            DeploymentUpdate(
                provider_data={
                    "connections": {},
                    "operations": [
                        {
                            "op": "unbind",
                            "tool": {"source_ref": str(uuid4()), "tool_id": str(uuid4())},
                            "app_ids": [str(donor_config_id)],
                        }
                    ],
                }
            ),
        )
        results.append(
            self._build_result(
                name="upd_reject_unbind_unknown_tool_id",
                expected={OUTCOME_INVALID_CONTENT},
                actual_outcome=status_code,
                detail=detail,
                ok=status_code == OUTCOME_INVALID_CONTENT,
            )
        )

        print("[upd/11] upd_missing_add_id_fails")
        status_code, detail, _ = await self._run_update(
            primary_deployment_id,
            DeploymentUpdate(
                provider_data={
                    "tools": {},
                    "connections": {},
                    "operations": [
                        {
                            "op": "bind",
                            "tool": {"tool_id_with_ref": {"source_ref": str(uuid4()), "tool_id": str(uuid4())}},
                            "app_ids": [str(donor_config_id)],
                        }
                    ],
                }
            ),
        )
        results.append(
            self._build_result(
                name="upd_missing_add_id_fails",
                expected={OUTCOME_INVALID_CONTENT},
                actual_outcome=status_code,
                detail=detail,
                ok=status_code == OUTCOME_INVALID_CONTENT,
            )
        )

        print("[upd/12] upd_config_raw_payload_conflict")
        conflict_seed_deployment_id, _conflict_cfg_id, _conflict_snapshot_ids, _ = await self._create_update_seed(
            label="upd_conflict_seed",
            snapshot_count=1,
        )
        conflict_suffix = uuid4().hex[:8]
        conflict_name = f"dup_cfg_{conflict_suffix}"
        conflict_tool_id = next(iter(_conflict_snapshot_ids), "")
        if not conflict_tool_id:
            results.append(
                self._build_result(
                    name="upd_config_raw_payload_conflict",
                    expected={OUTCOME_CONFLICT},
                    actual_outcome=OUTCOME_FAILURE,
                    detail="conflict seed snapshot id is missing",
                    ok=False,
                )
            )
            return results
        conflict_payload = DeploymentUpdate(
            provider_data={
                "tools": {},
                "connections": {"raw_payloads": [{"app_id": conflict_name, "environment_variables": {}}]},
                "operations": [
                    {
                        "op": "bind",
                        "tool": self._make_tool_id_with_ref(conflict_tool_id),
                        "app_ids": [conflict_name],
                    }
                ],
            }
        )
        setup_status, _setup_detail, setup_result = await self._run_update(
            conflict_seed_deployment_id, conflict_payload
        )
        setup_created_app_ids = self._extract_update_created_app_ids(setup_result)
        status_code, detail, _ = await self._run_update(conflict_seed_deployment_id, conflict_payload)
        results.append(
            self._build_result(
                name="upd_config_raw_payload_conflict",
                expected={OUTCOME_CONFLICT},
                actual_outcome=status_code,
                detail=(
                    f"setup={setup_status}:{_setup_detail} setup_created_app_ids={sorted(setup_created_app_ids)} "
                    f"conflict={status_code}:{detail}"
                ),
                ok=(
                    setup_status == OUTCOME_SUCCESS and bool(setup_created_app_ids) and status_code == OUTCOME_CONFLICT
                ),
            )
        )

        print("[upd/13] upd_not_found_deployment")
        status_code, detail, _ = await self._run_update(
            str(uuid4()),
            DeploymentUpdate(spec=BaseDeploymentDataUpdate(description="not found update")),
        )
        results.append(
            self._build_result(
                name="upd_not_found_deployment",
                expected={OUTCOME_NOT_FOUND},
                actual_outcome=status_code,
                detail=detail,
                ok=status_code == OUTCOME_NOT_FOUND,
            )
        )

        print("[upd/14] upd_put_tools_replaces_tool_list")
        put_tools_id, _put_tools_cfg, put_tools_snaps, _ = await self._create_update_seed(
            label="upd_put_tools", snapshot_count=2
        )
        put_tools_sorted = sorted(put_tools_snaps)
        keep_only = [put_tools_sorted[0]]
        status_code, detail, _ = await self._run_update(
            put_tools_id,
            DeploymentUpdate(provider_data={"put_tools": keep_only}),
        )
        list_status, _list_detail, snap_after = await self._run_list_snapshots(put_tools_id)
        attached_after = self._extract_snapshot_ids(snap_after)
        results.append(
            self._build_result(
                name="upd_put_tools_replaces_tool_list",
                expected={OUTCOME_SUCCESS},
                actual_outcome=status_code,
                detail=f"{detail} | attached_after={sorted(attached_after)} keep_only={keep_only}",
                ok=(
                    status_code == OUTCOME_SUCCESS
                    and list_status == OUTCOME_SUCCESS
                    and attached_after == set(keep_only)
                ),
            )
        )

        print("[upd/15] upd_put_tools_empty_clears_all_tools")
        status_code, detail, _ = await self._run_update(
            put_tools_id,
            DeploymentUpdate(provider_data={"put_tools": []}),
        )
        list_status, _list_detail, snap_after_clear = await self._run_list_snapshots(put_tools_id)
        attached_after_clear = self._extract_snapshot_ids(snap_after_clear)
        results.append(
            self._build_result(
                name="upd_put_tools_empty_clears_all_tools",
                expected={OUTCOME_SUCCESS},
                actual_outcome=status_code,
                detail=f"{detail} | attached_after={sorted(attached_after_clear)}",
                ok=(
                    status_code == OUTCOME_SUCCESS and list_status == OUTCOME_SUCCESS and len(attached_after_clear) == 0
                ),
            )
        )

        print("[upd/16] upd_put_tools_deduplicates")
        dup_id = put_tools_sorted[0]
        status_code, detail, _ = await self._run_update(
            put_tools_id,
            DeploymentUpdate(provider_data={"put_tools": [dup_id, dup_id, dup_id]}),
        )
        list_status, _list_detail, snap_after_dedup = await self._run_list_snapshots(put_tools_id)
        attached_after_dedup = self._extract_snapshot_ids(snap_after_dedup)
        results.append(
            self._build_result(
                name="upd_put_tools_deduplicates",
                expected={OUTCOME_SUCCESS},
                actual_outcome=status_code,
                detail=f"{detail} | attached_after={sorted(attached_after_dedup)}",
                ok=(
                    status_code == OUTCOME_SUCCESS
                    and list_status == OUTCOME_SUCCESS
                    and attached_after_dedup == {dup_id}
                ),
            )
        )

        # keep seed deployments tracked for shared final cleanup
        self.created_deployment_ids.add(primary_deployment_id)
        self.created_deployment_ids.add(donor_deployment_id)
        self.created_deployment_ids.add(mixed_donor_deployment_id)
        self.created_deployment_ids.add(put_tools_id)
        return results

    async def _run_live_concurrency_scenarios(self) -> list[ScenarioResult]:
        results: list[ScenarioResult] = []
        iterations_raw = os.getenv("WXO_CONCURRENCY_REPEAT", str(DEFAULT_CONCURRENCY_ITERATIONS)).strip()
        try:
            iterations = max(1, int(iterations_raw))
        except ValueError:
            iterations = DEFAULT_CONCURRENCY_ITERATIONS

        print(f"\n[concurrency] running iterations={iterations}")
        for iteration in range(1, iterations + 1):
            print(f"\n[concurrency] iteration {iteration}/{iterations}")
            results.extend(await self._run_live_concurrency_iteration(iteration=iteration))
        return results

    async def _run_live_concurrency_iteration(self, *, iteration: int) -> list[ScenarioResult]:
        results: list[ScenarioResult] = []

        print(f"[cc/{iteration}.1] cc_create_same_prefix_race")
        shared_dep_name = self._mk_name("dep_cc_shared")
        shared_cfg_name = self._mk_name("cfg_cc_shared")
        shared_snap_name = self._mk_name("snap_cc_shared")
        shared_payload = self._build_create_payload(
            tool_payloads=[self._build_flow_payload(label="cc_shared_snap", name_override=shared_snap_name)],
            raw_connection=DeploymentConfig(
                name=shared_cfg_name,
                description="concurrency create collision",
                environment_variables={},
            ),
        )
        shared_payload.spec = shared_payload.spec.model_copy(update={"name": shared_dep_name}, deep=True)
        create_race = await self._run_parallel_calls(
            {
                "left": lambda: self._run_create(shared_payload.model_copy(deep=True)),
                "right": lambda: self._run_create(shared_payload.model_copy(deep=True)),
            }
        )
        left_status, left_detail, left_created = create_race["left"]
        right_status, right_detail, right_created = create_race["right"]
        self._track_created_result(left_created)
        self._track_created_result(right_created)
        create_pair = (left_status, right_status)
        create_pair_ok = create_pair in {
            (OUTCOME_SUCCESS, OUTCOME_CONFLICT),
            (OUTCOME_CONFLICT, OUTCOME_SUCCESS),
            (OUTCOME_SUCCESS, OUTCOME_SUCCESS),
        }
        create_no_internal = OUTCOME_FAILURE not in {left_status, right_status}
        results.append(
            self._build_result(
                name="cc_create_same_prefix_race",
                expected={OUTCOME_SUCCESS, OUTCOME_CONFLICT},
                actual_outcome=max(left_status, right_status),
                detail=f"left={left_status}:{left_detail} right={right_status}:{right_detail}",
                ok=create_pair_ok and create_no_internal,
            )
        )

        print(f"[cc/{iteration}.2] cc_update_spec_vs_snapshot_race")
        primary_id, _primary_cfg_id, _primary_snaps, _ = await self._create_update_seed(
            label=f"cc_upd_primary_{iteration}",
            snapshot_count=2,
        )
        _donor_id, donor_cfg_id, donor_snapshot_ids, _ = await self._create_update_seed(
            label=f"cc_upd_donor_{iteration}",
            snapshot_count=1,
        )
        donor_snapshot_id = next(iter(donor_snapshot_ids), "")
        update_race = await self._run_parallel_calls(
            {
                "spec": lambda: self._run_update(
                    primary_id,
                    DeploymentUpdate(spec=BaseDeploymentDataUpdate(description="cc concurrent spec update")),
                ),
                "snapshot": lambda: self._run_update(
                    primary_id,
                    DeploymentUpdate(
                        provider_data={
                            "tools": {},
                            "connections": {},
                            "operations": [
                                {
                                    "op": "bind",
                                    "tool": self._make_tool_id_with_ref(donor_snapshot_id),
                                    "app_ids": [str(donor_cfg_id)],
                                }
                            ],
                        }
                    ),
                ),
            }
        )
        spec_status, spec_detail, _ = update_race["spec"]
        snapshot_status, snapshot_detail, _ = update_race["snapshot"]
        list_status, _list_detail, list_after = await self._run_list_snapshots(primary_id)
        attached_after = self._extract_snapshot_ids(list_after)
        no_internal_race = OUTCOME_FAILURE not in {spec_status, snapshot_status}
        results.append(
            self._build_result(
                name="cc_update_spec_vs_snapshot_race",
                expected={
                    OUTCOME_SUCCESS,
                    OUTCOME_INVALID_OPERATION,
                    OUTCOME_CONFLICT,
                    OUTCOME_INVALID_CONTENT,
                    OUTCOME_NOT_FOUND,
                },
                actual_outcome=max(spec_status, snapshot_status),
                detail=f"spec={spec_status}:{spec_detail} snapshot={snapshot_status}:{snapshot_detail}",
                ok=(
                    no_internal_race
                    and list_status == OUTCOME_SUCCESS
                    and self._has_unique_snapshot_ids(attached_after)
                ),
            )
        )

        print(f"[cc/{iteration}.3] cc_update_vs_delete_deployment_race")
        race_delete_id, _race_delete_cfg, _race_delete_snaps, _ = await self._create_update_seed(
            label=f"cc_upd_del_{iteration}",
            snapshot_count=1,
        )
        update_delete_race = await self._run_parallel_calls(
            {
                "update": lambda: self._run_update(
                    race_delete_id,
                    DeploymentUpdate(spec=BaseDeploymentDataUpdate(description="cc update while delete")),
                ),
                "delete": lambda: self._run_delete(race_delete_id),
            }
        )
        upd_status, upd_detail, _ = update_delete_race["update"]
        del_status, del_detail, _ = update_delete_race["delete"]
        status_after_delete_race, _status_detail, _status_payload = await self._run_status(race_delete_id)
        allowed_pairs = {
            (OUTCOME_SUCCESS, OUTCOME_SUCCESS),
            (OUTCOME_NOT_FOUND, OUTCOME_SUCCESS),
            (OUTCOME_SUCCESS, OUTCOME_NOT_FOUND),
            (OUTCOME_NOT_FOUND, OUTCOME_NOT_FOUND),
        }
        results.append(
            self._build_result(
                name="cc_update_vs_delete_deployment_race",
                expected={OUTCOME_SUCCESS, OUTCOME_NOT_FOUND, OUTCOME_FAILURE},
                actual_outcome=max(upd_status, del_status),
                detail=(
                    f"update={upd_status}:{upd_detail} "
                    f"delete={del_status}:{del_detail} status={status_after_delete_race}"
                ),
                ok=(upd_status, del_status) in allowed_pairs
                or (
                    OUTCOME_FAILURE in {upd_status, del_status} and "not found" in f"{upd_detail} {del_detail}".lower()
                ),
            )
        )
        self.created_deployment_ids.discard(race_delete_id)

        print(f"[cc/{iteration}.4] cc_execution_vs_delete_deployment_race")
        exec_delete_id, _exec_delete_cfg, _exec_delete_snapshots, _ = await self._create_update_seed(
            label=f"cc_exec_del_{iteration}",
            snapshot_count=1,
        )
        exec_delete_race = await self._run_parallel_calls(
            {
                "execution": lambda: self._run_create_execution(
                    exec_delete_id,
                    provider_data={"message": {"role": "user", "content": "woah"}},
                ),
                "delete": lambda: self._run_delete(exec_delete_id),
            }
        )
        exec_status, exec_detail, _ = exec_delete_race["execution"]
        del_exec_status, del_exec_detail, _ = exec_delete_race["delete"]
        results.append(
            self._build_result(
                name="cc_execution_vs_delete_deployment_race",
                expected={OUTCOME_SUCCESS, OUTCOME_NOT_FOUND, OUTCOME_SUCCESS, OUTCOME_FAILURE},
                actual_outcome=max(exec_status, del_exec_status),
                detail=f"execution={exec_status}:{exec_detail} delete={del_exec_status}:{del_exec_detail}",
                ok=OUTCOME_FAILURE not in {exec_status, del_exec_status}
                or "not found" in f"{exec_detail} {del_exec_detail}".lower(),
            )
        )
        self.created_deployment_ids.discard(exec_delete_id)

        print(f"[cc/{iteration}.5] cc_delete_snapshot_during_update_bindings")
        delete_bind_id, bind_cfg_id, bind_snapshot_ids, _ = await self._create_update_seed(
            label=f"cc_del_bind_{iteration}",
            snapshot_count=2,
        )
        delete_target_snapshot_id = next(iter(bind_snapshot_ids), "")

        async def _delete_target_before_bind(*args: Any, **kwargs: Any) -> None:  # noqa: ARG001
            clients = kwargs.get("clients")
            existing_tool_deltas = kwargs.get("existing_tool_deltas") or {}
            target_ids = list(existing_tool_deltas.keys())
            tool_id = str(target_ids[0]) if target_ids else delete_target_snapshot_id
            if clients and tool_id:
                await self._safe_delete_snapshot(clients=clients, snapshot_id=tool_id)

        del_bind_status, del_bind_detail, _ = await self._run_with_stage_hook(
            stage="update_bindings",
            operation=lambda: self._run_update(
                delete_bind_id,
                DeploymentUpdate(
                    provider_data={
                        "tools": {},
                        "connections": {},
                        "operations": [
                            {
                                "op": "unbind",
                                "tool": self._make_tool_ref(tool_id),
                                "app_ids": [str(bind_cfg_id)],
                            }
                            for tool_id in sorted(bind_snapshot_ids)
                        ],
                    }
                ),
            ),
            hook_before=_delete_target_before_bind,
        )
        results.append(
            self._build_result(
                name="cc_delete_snapshot_during_update_bindings",
                expected={OUTCOME_INVALID_CONTENT, OUTCOME_INVALID_OPERATION, OUTCOME_SUCCESS},
                actual_outcome=del_bind_status,
                detail=del_bind_detail,
                ok=del_bind_status in {OUTCOME_SUCCESS, OUTCOME_INVALID_OPERATION, OUTCOME_INVALID_CONTENT},
            )
        )

        print(f"[cc/{iteration}.6] cc_delete_config_after_update_raw_create")
        delete_cfg_id, _delete_cfg_base_id, delete_cfg_snapshots, _ = await self._create_update_seed(
            label=f"cc_del_cfg_{iteration}",
            snapshot_count=1,
        )
        target_tool_id = next(iter(delete_cfg_snapshots), "")
        if not target_tool_id:
            results.append(
                self._build_result(
                    name="cc_delete_config_after_update_raw_create",
                    expected={OUTCOME_INVALID_OPERATION, OUTCOME_INVALID_CONTENT, OUTCOME_CONFLICT, OUTCOME_FAILURE},
                    actual_outcome=OUTCOME_FAILURE,
                    detail="seed snapshot id missing for update raw config stage",
                    ok=False,
                )
            )
            return results
        raw_cfg_name = self._mk_name("cc_raw_cfg")

        async def _delete_created_app_after_config_create(created_app_id: Any, **kwargs: Any) -> None:
            if not created_app_id:
                created_app_id = kwargs.get("app_id")
            app_id = str(created_app_id or "").strip()
            clients = await self.service._get_provider_clients(user_id=self.user_id, db=self.db)  # noqa: SLF001
            if app_id:
                await self._safe_delete_config(clients=clients, config_id=app_id)

        del_cfg_status, del_cfg_detail, _ = await self._run_with_stage_hook(
            stage="update_create_config",
            operation=lambda: self._run_update(
                delete_cfg_id,
                DeploymentUpdate(
                    provider_data={
                        "tools": {},
                        "connections": {"raw_payloads": [{"app_id": raw_cfg_name, "environment_variables": {}}]},
                        "operations": [
                            {
                                "op": "bind",
                                "tool": self._make_tool_id_with_ref(target_tool_id),
                                "app_ids": [raw_cfg_name],
                            }
                        ],
                    }
                ),
            ),
            hook_after=_delete_created_app_after_config_create,
        )
        results.append(
            self._build_result(
                name="cc_delete_config_after_update_raw_create",
                expected={OUTCOME_INVALID_OPERATION, OUTCOME_INVALID_CONTENT, OUTCOME_CONFLICT, OUTCOME_FAILURE},
                actual_outcome=del_cfg_status,
                detail=del_cfg_detail,
                ok=del_cfg_status
                in {
                    OUTCOME_INVALID_OPERATION,
                    OUTCOME_INVALID_CONTENT,
                    OUTCOME_CONFLICT,
                    OUTCOME_FAILURE,
                },
            )
        )

        print(f"[cc/{iteration}.7] cc_create_during_create_snapshots_stage")
        create_race_dep = self._mk_name("cc_stage_dep")
        create_race_cfg = self._mk_name("cc_stage_cfg")
        create_race_snap = self._mk_name("cc_stage_snap")
        race_payload = self._build_create_payload(
            tool_payloads=[self._build_flow_payload(label="cc_stage_snap", name_override=create_race_snap)],
            raw_connection=DeploymentConfig(
                name=create_race_cfg,
                description="cc competing create",
                environment_variables={},
            ),
        )
        race_payload.spec = race_payload.spec.model_copy(update={"name": create_race_dep}, deep=True)
        competing_create_task: asyncio.Task[tuple[str, str, WxoCreatedDeploymentResult | None]] | None = None

        async def _launch_competing_create(*args: Any, **kwargs: Any) -> None:  # noqa: ARG001
            nonlocal competing_create_task
            if competing_create_task is None:
                competing_create_task = asyncio.create_task(self._run_create(race_payload.model_copy(deep=True)))
                await asyncio.sleep(0)

        staged_create_status, staged_create_detail, staged_create_created = await self._run_with_stage_hook(
            stage="create_snapshots",
            operation=lambda: self._run_create(race_payload.model_copy(deep=True)),
            hook_before=_launch_competing_create,
        )
        competing_create_result = (OUTCOME_FAILURE, "competing create did not start", None)
        if competing_create_task is not None:
            competing_create_result = await competing_create_task
        comp_status, comp_detail, comp_created = competing_create_result
        self._track_created_result(staged_create_created)
        self._track_created_result(comp_created)
        staged_pair = (staged_create_status, comp_status)
        results.append(
            self._build_result(
                name="cc_create_during_create_snapshots_stage",
                expected={OUTCOME_SUCCESS, OUTCOME_CONFLICT},
                actual_outcome=max(staged_create_status, comp_status),
                detail=f"main={staged_create_status}:{staged_create_detail} competing={comp_status}:{comp_detail}",
                ok=(
                    staged_pair
                    in {
                        (OUTCOME_SUCCESS, OUTCOME_CONFLICT),
                        (OUTCOME_CONFLICT, OUTCOME_SUCCESS),
                        (OUTCOME_SUCCESS, OUTCOME_SUCCESS),
                    }
                    and OUTCOME_FAILURE not in staged_pair
                ),
            )
        )

        print(f"[cc/{iteration}.8] cc_create_during_update_raw_config_stage")
        update_create_id, _update_create_cfg, update_create_snaps, _ = await self._create_update_seed(
            label=f"cc_create_update_cfg_{iteration}",
            snapshot_count=1,
        )
        update_target_tool_id = next(iter(update_create_snaps), "")
        if not update_target_tool_id:
            results.append(
                self._build_result(
                    name="cc_create_during_update_raw_config_stage",
                    expected={OUTCOME_SUCCESS, OUTCOME_CONFLICT},
                    actual_outcome=OUTCOME_FAILURE,
                    detail="seed snapshot id missing for competing update",
                    ok=False,
                )
            )
            return results
        update_cfg_name = self._mk_name("cc_upd_cfg_create")
        competing_update_task: asyncio.Task[tuple[str, str, Any | None]] | None = None

        competing_update_payload = DeploymentUpdate(
            provider_data={
                "tools": {},
                "connections": {"raw_payloads": [{"app_id": update_cfg_name, "environment_variables": {}}]},
                "operations": [
                    {
                        "op": "bind",
                        "tool": self._make_tool_id_with_ref(update_target_tool_id),
                        "app_ids": [update_cfg_name],
                    }
                ],
            },
        )

        async def _launch_competing_update_create(*args: Any, **kwargs: Any) -> None:  # noqa: ARG001
            nonlocal competing_update_task
            if competing_update_task is None:
                competing_update_task = asyncio.create_task(
                    self._run_update(update_create_id, competing_update_payload.model_copy(deep=True))
                )
                await asyncio.sleep(0)

        update_cfg_status, update_cfg_detail, _ = await self._run_with_stage_hook(
            stage="update_create_config",
            operation=lambda: self._run_update(update_create_id, competing_update_payload.model_copy(deep=True)),
            hook_before=_launch_competing_update_create,
        )
        competing_update_result = (OUTCOME_FAILURE, "competing update did not start", None)
        if competing_update_task is not None:
            competing_update_result = await competing_update_task
        competing_upd_status, competing_upd_detail, _ = competing_update_result
        results.append(
            self._build_result(
                name="cc_create_during_update_raw_config_stage",
                expected={OUTCOME_SUCCESS, OUTCOME_CONFLICT},
                actual_outcome=max(update_cfg_status, competing_upd_status),
                detail=(
                    f"main={update_cfg_status}:{update_cfg_detail} "
                    f"competing={competing_upd_status}:{competing_upd_detail}"
                ),
                ok=OUTCOME_FAILURE not in {update_cfg_status, competing_upd_status},
            )
        )

        print(f"[cc/{iteration}.9] cc_delete_resources_during_update_rollback")
        rollback_id, rollback_cfg_id, rollback_seed_snapshots, _ = await self._create_update_seed(
            label=f"cc_rollback_delete_{iteration}",
            snapshot_count=1,
        )
        if not rollback_cfg_id:
            results.append(
                self._build_result(
                    name="cc_delete_resources_during_update_rollback",
                    expected={OUTCOME_SUCCESS},
                    actual_outcome=OUTCOME_FAILURE,
                    detail="seed rollback config id missing",
                    ok=False,
                )
            )
            return results
        rollback_seed_tool_id = next(iter(rollback_seed_snapshots), "")
        if not rollback_seed_tool_id:
            results.append(
                self._build_result(
                    name="cc_delete_resources_during_update_rollback",
                    expected={OUTCOME_FAILURE},
                    actual_outcome=OUTCOME_FAILURE,
                    detail="seed snapshot id missing for rollback race",
                    ok=False,
                )
            )
            return results
        rollback_raw_flow = self._build_flow_payload(label=f"cc_rb_raw_{iteration}")
        rollback_raw_cfg_name = self._mk_name("cc_rb_cfg")
        rollback_status, rollback_detail, _ = await self._run_with_stage_hook(
            stage="update_rollback_resources",
            operation=lambda: self._run_update(
                rollback_id,
                DeploymentUpdate(
                    spec=BaseDeploymentDataUpdate(description="cc rollback delete race"),
                    provider_data={
                        "tools": {
                            "raw_payloads": [rollback_raw_flow.model_dump(mode="json")],
                        },
                        "connections": {
                            "raw_payloads": [{"app_id": rollback_raw_cfg_name, "environment_variables": {}}],
                        },
                        "operations": [
                            {
                                "op": "unbind",
                                "tool": self._make_tool_ref(rollback_seed_tool_id),
                                "app_ids": [str(rollback_cfg_id)],
                            },
                            {
                                "op": "bind",
                                "tool": {"name_of_raw": rollback_raw_flow.name},
                                "app_ids": [rollback_raw_cfg_name],
                            },
                        ],
                    },
                ),
                inject={
                    "update_bindings": {
                        "fail_first_n": 1,
                        "error_type": "runtime",
                        "message": "cc_rollback_trigger",
                    }
                },
            ),
            hook_before=self._delete_resources_before_rollback_hook,
        )
        results.append(
            self._build_result(
                name="cc_delete_resources_during_update_rollback",
                expected={OUTCOME_FAILURE},
                actual_outcome=rollback_status,
                detail=rollback_detail,
                ok=rollback_status == OUTCOME_FAILURE,
            )
        )

        print(f"[cc/{iteration}.10] cc_create_during_update_rollback")
        rollback_create_id, rollback_create_cfg_id, rollback_create_snaps, _ = await self._create_update_seed(
            label=f"cc_rollback_create_{iteration}",
            snapshot_count=1,
        )
        if not rollback_create_cfg_id:
            results.append(
                self._build_result(
                    name="cc_create_during_update_rollback",
                    expected={OUTCOME_SUCCESS},
                    actual_outcome=OUTCOME_FAILURE,
                    detail="seed rollback-create config id missing",
                    ok=False,
                )
            )
            return results
        rollback_create_seed_tool_id = next(iter(rollback_create_snaps), "")
        if not rollback_create_seed_tool_id:
            results.append(
                self._build_result(
                    name="cc_create_during_update_rollback",
                    expected={OUTCOME_FAILURE, OUTCOME_SUCCESS, OUTCOME_CONFLICT},
                    actual_outcome=OUTCOME_FAILURE,
                    detail="seed snapshot id missing for rollback create race",
                    ok=False,
                )
            )
            return results
        rollback_create_cfg_name = self._mk_name("cc_rb_create_cfg")
        competing_rollback_create_task: asyncio.Task[tuple[str, str, Any | None]] | None = None

        async def _launch_competing_create_before_rollback(*args: Any, **kwargs: Any) -> None:  # noqa: ARG001
            nonlocal competing_rollback_create_task
            if competing_rollback_create_task is not None:
                return
            competing_rollback_create_task = asyncio.create_task(
                self._run_update(
                    rollback_create_id,
                    DeploymentUpdate(
                        provider_data={
                            "tools": {},
                            "connections": {
                                "raw_payloads": [{"app_id": rollback_create_cfg_name, "environment_variables": {}}]
                            },
                            "operations": [
                                {
                                    "op": "bind",
                                    "tool": self._make_tool_id_with_ref(rollback_create_seed_tool_id),
                                    "app_ids": [rollback_create_cfg_name],
                                }
                            ],
                        },
                    ),
                )
            )
            await asyncio.sleep(0)

        rollback_create_raw_flow = self._build_flow_payload(label=f"cc_rb_create_raw_{iteration}")
        rollback_create_status, rollback_create_detail, _ = await self._run_with_stage_hook(
            stage="update_rollback_resources",
            operation=lambda: self._run_update(
                rollback_create_id,
                DeploymentUpdate(
                    spec=BaseDeploymentDataUpdate(description="cc rollback create race"),
                    provider_data={
                        "tools": {
                            "raw_payloads": [rollback_create_raw_flow.model_dump(mode="json")],
                        },
                        "connections": {
                            "raw_payloads": [{"app_id": rollback_create_cfg_name, "environment_variables": {}}],
                        },
                        "operations": [
                            {
                                "op": "unbind",
                                "tool": self._make_tool_ref(rollback_create_seed_tool_id),
                                "app_ids": [str(rollback_create_cfg_id)],
                            },
                            {
                                "op": "bind",
                                "tool": {"name_of_raw": rollback_create_raw_flow.name},
                                "app_ids": [rollback_create_cfg_name],
                            },
                        ],
                    },
                ),
                inject={
                    "update_bindings": {
                        "fail_first_n": 1,
                        "error_type": "runtime",
                        "message": "cc_rollback_create_trigger",
                    }
                },
            ),
            hook_before=_launch_competing_create_before_rollback,
        )
        competing_rollback_create_result = (OUTCOME_FAILURE, "competing rollback create missing", None)
        if competing_rollback_create_task is not None:
            competing_rollback_create_result = await competing_rollback_create_task
        comp_rb_status, comp_rb_detail, _ = competing_rollback_create_result
        results.append(
            self._build_result(
                name="cc_create_during_update_rollback",
                expected={OUTCOME_FAILURE, OUTCOME_SUCCESS, OUTCOME_CONFLICT},
                actual_outcome=max(rollback_create_status, comp_rb_status),
                detail=(
                    f"main={rollback_create_status}:{rollback_create_detail} "
                    f"competing={comp_rb_status}:{comp_rb_detail}"
                ),
                ok=rollback_create_status == OUTCOME_FAILURE
                and comp_rb_status in {OUTCOME_SUCCESS, OUTCOME_CONFLICT, OUTCOME_FAILURE},
            )
        )

        print(f"[cc/{iteration}.11] cc_parallel_updates_isolation")
        dep_a, cfg_a, _snap_a, _ = await self._create_update_seed(label=f"cc_iso_a_{iteration}", snapshot_count=1)
        dep_b, cfg_b, _snap_b, _ = await self._create_update_seed(label=f"cc_iso_b_{iteration}", snapshot_count=1)
        isolation_race = await self._run_parallel_calls(
            {
                "a": lambda: self._run_update(
                    dep_a,
                    DeploymentUpdate(spec=BaseDeploymentDataUpdate(description="isolation-a")),
                ),
                "b": lambda: self._run_update(
                    dep_b,
                    DeploymentUpdate(spec=BaseDeploymentDataUpdate(description="isolation-b")),
                ),
            }
        )
        iso_a_status, iso_a_detail, _ = isolation_race["a"]
        iso_b_status, iso_b_detail, _ = isolation_race["b"]
        cfg_list_a_status, _cfg_a_detail, cfg_list_a = await self._run_list_configs(dep_a)
        cfg_list_b_status, _cfg_b_detail, cfg_list_b = await self._run_list_configs(dep_b)
        cfg_ids_a = self._extract_config_ids(cfg_list_a)
        cfg_ids_b = self._extract_config_ids(cfg_list_b)
        print(
            "[cc/debug] parallel_updates_isolation "
            f"dep_a={dep_a} cfg_a={cfg_a} update_a={iso_a_status}:{iso_a_detail} "
            f"list_a={cfg_list_a_status}:{_cfg_a_detail} cfg_ids_a={sorted(cfg_ids_a)} "
            f"dep_b={dep_b} cfg_b={cfg_b} update_b={iso_b_status}:{iso_b_detail} "
            f"list_b={cfg_list_b_status}:{_cfg_b_detail} cfg_ids_b={sorted(cfg_ids_b)}"
        )
        isolation_ok = (
            cfg_list_a_status == OUTCOME_SUCCESS and cfg_list_b_status == OUTCOME_SUCCESS and cfg_ids_a and cfg_ids_b
        )
        if cfg_a:
            isolation_ok = isolation_ok and str(cfg_a) in cfg_ids_a
        if cfg_b:
            isolation_ok = isolation_ok and str(cfg_b) in cfg_ids_b
        results.append(
            self._build_result(
                name="cc_parallel_updates_isolation",
                expected={OUTCOME_SUCCESS},
                actual_outcome=max(iso_a_status, iso_b_status),
                detail=f"a={iso_a_status}:{iso_a_detail} b={iso_b_status}:{iso_b_detail}",
                ok=isolation_ok and iso_a_status == OUTCOME_SUCCESS and iso_b_status == OUTCOME_SUCCESS,
            )
        )
        return results

    async def _run_parallel_calls(
        self,
        calls: dict[str, Callable[[], Awaitable[tuple[str, str, Any | None]]]],
    ) -> dict[str, tuple[str, str, Any | None]]:
        tasks = {name: asyncio.create_task(call()) for name, call in calls.items()}
        gathered = await asyncio.gather(*tasks.values(), return_exceptions=True)
        results: dict[str, tuple[str, str, Any | None]] = {}
        for name, outcome in zip(tasks, gathered, strict=False):
            if isinstance(outcome, Exception):
                results[name] = (OUTCOME_FAILURE, str(outcome), None)
            else:
                results[name] = outcome
        return results

    def _track_created_result(self, created: WxoCreatedDeploymentResult | None) -> None:
        if not created:
            return
        self.created_deployment_ids.add(created.deployment_id)
        self.created_config_ids.update(self._extract_create_app_ids(created.provider_result))
        self.created_snapshot_ids.update(self._extract_create_snapshot_ids(created.provider_result))

    def _has_unique_snapshot_ids(self, snapshot_ids: set[str]) -> bool:
        snapshot_list = [str(item) for item in snapshot_ids]
        return len(snapshot_list) == len(set(snapshot_list))

    def _extract_config_ids(self, config_result: Any) -> set[str]:
        configs = getattr(config_result, "configs", []) if config_result else []
        ids_or_names: set[str] = set()
        for config in configs:
            if not config:
                continue
            config_id = str(getattr(config, "id", "")).strip()
            config_name = str(getattr(config, "name", "")).strip()
            if config_id:
                ids_or_names.add(config_id)
            if config_name:
                ids_or_names.add(config_name)
        return ids_or_names

    def _stage_hook_mapping(self) -> dict[str, tuple[Any, str]]:
        return {
            "create_config": (create_core_module, "create_connection_with_conflict_mapping"),
            "create_snapshots": (create_core_module, "create_and_upload_wxo_flow_tools_with_bindings"),
            "create_agent": (create_core_module, "create_agent_deployment"),
            "update_create_config": (update_core_module, "create_connection_with_conflict_mapping"),
            "update_create_tools": (update_core_module, "create_and_upload_wxo_flow_tools_with_bindings"),
            "update_bindings": (update_core_module, "_update_existing_tool_connection_deltas"),
            "update_rollback_resources": (update_core_module, "rollback_update_resources"),
        }

    async def _run_with_stage_hook(
        self,
        *,
        stage: str,
        operation: Callable[[], Awaitable[tuple[str, str, Any | None]]],
        hook_before: Callable[..., Awaitable[None]] | None = None,
        hook_after: Callable[..., Awaitable[None]] | None = None,
    ) -> tuple[str, str, Any | None]:
        originals: list[tuple[Any, str, Any]] = []
        self._apply_stage_hook(
            stage=stage,
            originals=originals,
            hook_before=hook_before,
            hook_after=hook_after,
        )
        try:
            return await operation()
        finally:
            for target, attr_name, original in originals:
                setattr(target, attr_name, original)

    def _apply_stage_hook(
        self,
        *,
        stage: str,
        originals: list[tuple[Any, str, Any]],
        hook_before: Callable[..., Awaitable[None]] | None = None,
        hook_after: Callable[..., Awaitable[None]] | None = None,
    ) -> None:
        mapping = self._stage_hook_mapping()
        target_and_method = mapping.get(stage)
        if target_and_method is None:
            return
        target, method_name = target_and_method
        original = getattr(target, method_name)
        originals.append((target, method_name, original))
        before_called = {"value": False}
        after_called = {"value": False}

        if isinstance(target, types.ModuleType):

            async def _wrapped_module(
                *args,
                __orig=original,
                __before=hook_before,
                __after=hook_after,
                __before_called=before_called,
                __after_called=after_called,
                **kwargs,
            ):
                if __before is not None and not __before_called["value"]:
                    __before_called["value"] = True
                    await __before(*args, **kwargs)
                result = await __orig(*args, **kwargs)
                if __after is not None and not __after_called["value"]:
                    __after_called["value"] = True
                    await __after(result, *args, **kwargs)
                return result

            setattr(target, method_name, _wrapped_module)
            return

        async def _wrapped_method(
            _self,
            *args,
            __orig=original,
            __before=hook_before,
            __after=hook_after,
            __before_called=before_called,
            __after_called=after_called,
            **kwargs,
        ):
            if __before is not None and not __before_called["value"]:
                __before_called["value"] = True
                await __before(*args, **kwargs)
            result = await __orig(*args, **kwargs)
            if __after is not None and not __after_called["value"]:
                __after_called["value"] = True
                await __after(result, *args, **kwargs)
            return result

        setattr(target, method_name, MethodType(_wrapped_method, target))

    async def _safe_delete_snapshot(self, *, clients: Any, snapshot_id: str) -> None:
        with suppress(ClientAPIException):
            await asyncio.to_thread(clients.tool.delete, snapshot_id)

    async def _safe_delete_config(self, *, clients: Any, config_id: str) -> None:
        with suppress(ClientAPIException):
            await asyncio.to_thread(clients.connections.delete, config_id)

    async def _delete_resources_before_rollback_hook(self, *args: Any, **kwargs: Any) -> None:  # noqa: ARG002
        clients = kwargs.get("clients")
        if clients is None:
            return
        for tool_id in kwargs.get("created_tool_ids") or []:
            await self._safe_delete_snapshot(clients=clients, snapshot_id=str(tool_id))
        app_id = kwargs.get("created_app_id")
        if app_id:
            await self._safe_delete_config(clients=clients, config_id=str(app_id))
        for created_app_id in kwargs.get("created_app_ids") or []:
            await self._safe_delete_config(clients=clients, config_id=str(created_app_id))

    async def _run_update_failpoint_scenarios(self) -> list[ScenarioResult]:
        results: list[ScenarioResult] = []
        print("\n[fp-upd] creating seed deployment")
        deployment_id, config_id, snapshot_ids, _ = await self._create_update_seed(
            label="fp_upd_seed",
            snapshot_count=1,
        )
        if not config_id:
            results.append(
                self._build_result(
                    name="fp_update_seed_missing_config",
                    expected={OUTCOME_SUCCESS},
                    actual_outcome=OUTCOME_FAILURE,
                    detail="seed deployment config id is missing",
                    ok=False,
                )
            )
            return results
        seed_tool_id = next(iter(snapshot_ids), "")
        if not seed_tool_id:
            results.append(
                self._build_result(
                    name="fp_update_seed_missing_snapshot",
                    expected={OUTCOME_SUCCESS},
                    actual_outcome=OUTCOME_FAILURE,
                    detail="seed deployment snapshot id is missing",
                    ok=False,
                )
            )
            return results
        failpoint_raw_app_id = self._mk_name("fp_upd_cfg")

        update_payload = DeploymentUpdate(
            spec=BaseDeploymentDataUpdate(description="trigger update failpoint"),
            provider_data={
                "tools": {},
                "connections": {"raw_payloads": [{"app_id": failpoint_raw_app_id, "environment_variables": {}}]},
                "operations": [
                    {
                        "op": "bind",
                        "tool": self._make_tool_id_with_ref(seed_tool_id),
                        "app_ids": [failpoint_raw_app_id],
                    }
                ],
            },
        )

        print("[fp-upd/1] fp_update_bindings_failure_triggers_rollback")
        status_code, detail, _ = await self._run_update(
            deployment_id,
            update_payload,
            inject={
                "update_bindings": {
                    "fail_first_n": 1,
                    "error_type": "runtime",
                    "message": "fp_update_bindings_failure",
                }
            },
        )
        results.append(
            self._build_result(
                name="fp_update_bindings_failure_triggers_rollback",
                expected={OUTCOME_FAILURE},
                actual_outcome=status_code,
                detail=detail,
                ok=status_code == OUTCOME_FAILURE,
            )
        )

        print("[fp-upd/2] fp_update_bindings_failure_with_rollback_failure")
        status_code, detail, _ = await self._run_update(
            deployment_id,
            update_payload,
            inject={
                "update_bindings": {
                    "fail_first_n": 1,
                    "error_type": "runtime",
                    "message": "fp_update_bindings_failure_again",
                },
                "update_rollback_resources": {
                    "fail_first_n": 1,
                    "error_type": "runtime",
                    "message": "fp_update_rollback_failure",
                },
            },
        )
        results.append(
            self._build_result(
                name="fp_update_bindings_failure_with_rollback_failure",
                expected={OUTCOME_FAILURE},
                actual_outcome=status_code,
                detail=detail,
                ok=status_code == OUTCOME_FAILURE,
            )
        )

        print("[fp-upd/3] fp_update_failure_then_put_tools_restore")
        restore_id, restore_cfg_id, restore_snaps, _ = await self._create_update_seed(
            label="fp_put_tools_restore",
            snapshot_count=2,
        )
        original_snap_ids = sorted(restore_snaps)
        if not restore_cfg_id or len(original_snap_ids) < MIN_MIXED_SNAPSHOT_IDS:
            results.append(
                self._build_result(
                    name="fp_update_failure_then_put_tools_restore",
                    expected={OUTCOME_SUCCESS},
                    actual_outcome=OUTCOME_FAILURE,
                    detail=f"seed insufficient: cfg={restore_cfg_id} snaps={len(original_snap_ids)}",
                    ok=False,
                )
            )
            return results

        restore_raw_cfg = self._mk_name("fp_restore_cfg")
        print("[fp-upd/3a] injecting update failure to corrupt tool list")
        inject_status, inject_detail, _ = await self._run_update(
            restore_id,
            DeploymentUpdate(
                provider_data={
                    "tools": {},
                    "connections": {"raw_payloads": [{"app_id": restore_raw_cfg, "environment_variables": {}}]},
                    "operations": [
                        {
                            "op": "bind",
                            "tool": self._make_tool_id_with_ref(original_snap_ids[0]),
                            "app_ids": [restore_raw_cfg],
                        }
                    ],
                }
            ),
            inject={
                "update_bindings": {
                    "fail_first_n": 1,
                    "error_type": "runtime",
                    "message": "fp_put_tools_restore_trigger",
                }
            },
        )
        failure_triggered = inject_status == OUTCOME_FAILURE

        print("[fp-upd/3b] restoring via put_tools")
        restore_status, restore_detail, _ = await self._run_update(
            restore_id,
            DeploymentUpdate(provider_data={"put_tools": original_snap_ids}),
        )
        list_status, _list_detail, snap_after_restore = await self._run_list_snapshots(restore_id)
        attached_after_restore = self._extract_snapshot_ids(snap_after_restore)
        restored_ok = set(original_snap_ids) == attached_after_restore
        results.append(
            self._build_result(
                name="fp_update_failure_then_put_tools_restore",
                expected={OUTCOME_SUCCESS},
                actual_outcome=restore_status,
                detail=(
                    f"failure_triggered={failure_triggered} inject={inject_status}:{inject_detail} "
                    f"restore={restore_status}:{restore_detail} "
                    f"attached_after={sorted(attached_after_restore)} expected={original_snap_ids}"
                ),
                ok=(
                    failure_triggered
                    and restore_status == OUTCOME_SUCCESS
                    and list_status == OUTCOME_SUCCESS
                    and restored_ok
                ),
            )
        )

        return results

    def _apply_injections(self, inject: dict[str, dict[str, Any]], originals: list[tuple[Any, str, Any]]) -> None:
        mapping = {
            "create_config": (shared_core_module, "create_config"),
            "create_config_wrapper": (create_core_module, "create_connection_with_conflict_mapping"),
            "create_snapshots": (create_core_module, "create_and_upload_wxo_flow_tools_with_bindings"),
            "create_agent": (create_core_module, "create_agent_deployment"),
            "rollback_delete_agent": (retry_module, "delete_agent_if_exists"),
            "rollback_delete_tool": (retry_module, "delete_tool_if_exists"),
            "rollback_delete_config": (retry_module, "delete_config_if_exists"),
            "update_bindings": (update_core_module, "_update_existing_tool_connection_deltas"),
            "update_rollback_resources": (update_core_module, "rollback_update_resources"),
        }

        for stage, config in inject.items():
            target_and_method = mapping.get(stage)
            if target_and_method is None:
                continue
            target, method_name = target_and_method
            original = getattr(target, method_name)
            originals.append((target, method_name, original))
            counter = {"value": 0}
            fail_first_n = int(config.get("fail_first_n", 0))
            error_type = str(config.get("error_type", "runtime")).strip().lower()
            message = str(config.get("message") or f"injected failure: {stage}")
            if isinstance(target, types.ModuleType):

                async def _wrapped_module(
                    *args,
                    __orig=original,
                    __ctr=counter,
                    __n=fail_first_n,
                    __type=error_type,
                    __msg=message,
                    **kwargs,
                ):
                    __ctr["value"] += 1
                    if __ctr["value"] <= __n:
                        if __type == "domain_conflict":
                            raise ResourceConflictError(message=__msg)
                        if __type == "domain_not_found":
                            raise DeploymentNotFoundError(message=__msg)
                        if __type == "domain_invalid_content":
                            raise InvalidContentError(message=__msg)
                        if __type == "domain_invalid_operation":
                            raise InvalidDeploymentOperationError(message=__msg)
                        if __type == "domain_failure":
                            raise DeploymentError(message=__msg, error_code="deployment_error")
                        raise RuntimeError(__msg)
                    return await __orig(*args, **kwargs)

                setattr(target, method_name, _wrapped_module)
                continue

            async def _wrapped_method(
                _self,
                *args,
                __orig=original,
                __ctr=counter,
                __n=fail_first_n,
                __type=error_type,
                __msg=message,
                **kwargs,
            ):
                __ctr["value"] += 1
                if __ctr["value"] <= __n:
                    if __type == "domain_conflict":
                        raise ResourceConflictError(message=__msg)
                    if __type == "domain_not_found":
                        raise DeploymentNotFoundError(message=__msg)
                    if __type == "domain_invalid_content":
                        raise InvalidContentError(message=__msg)
                    if __type == "domain_invalid_operation":
                        raise InvalidDeploymentOperationError(message=__msg)
                    if __type == "domain_failure":
                        raise DeploymentError(message=__msg, error_code="deployment_error")
                    raise RuntimeError(__msg)
                return await __orig(*args, **kwargs)

            setattr(target, method_name, MethodType(_wrapped_method, target))

    def _build_create_payload(
        self,
        *,
        tool_payloads: list[BaseFlowArtifact[WatsonxFlowArtifactProviderData]],
        raw_connection: DeploymentConfig | None = None,
        existing_connection_app_id: str | None = None,
    ) -> DeploymentCreate:
        spec = BaseDeploymentData(
            name=self._mk_name("dep_agent"),
            description="direct adapter scenario",
            type=DeploymentType.AGENT,
        )
        raw_tool_payloads = [snapshot.model_copy(deep=True) for snapshot in tool_payloads]
        connections: dict[str, Any] = {}

        if raw_connection is not None:
            operation_app_id = str(raw_connection.name).strip()
            connections["raw_payloads"] = [
                {
                    "app_id": operation_app_id,
                    "environment_variables": raw_connection.environment_variables,
                    "provider_config": raw_connection.provider_config,
                }
            ]
        elif existing_connection_app_id:
            operation_app_id = str(existing_connection_app_id).strip()
        else:
            operation_app_id = self._mk_name("cfg_default_app")
            connections["raw_payloads"] = [{"app_id": operation_app_id, "environment_variables": {}}]

        operations = [
            {
                "op": "bind",
                "tool": {"name_of_raw": str(flow_payload.name).strip()},
                "app_ids": [operation_app_id],
            }
            for flow_payload in raw_tool_payloads
        ]

        provider_data = {
            "tools": {"raw_payloads": raw_tool_payloads},
            "connections": connections,
            "operations": operations,
            "llm": self.llm,
        }
        return DeploymentCreate(spec=spec, provider_data=provider_data)

    def _build_flow_payload(
        self,
        *,
        label: str,
        name_override: str | None = None,
    ) -> BaseFlowArtifact[WatsonxFlowArtifactProviderData]:
        flow_id = uuid4()
        return BaseFlowArtifact[WatsonxFlowArtifactProviderData](
            id=flow_id,
            name=name_override or self._mk_name(label),
            description="direct adapter flow payload",
            data=self._build_flow_data_payload(),
            tags=["e2e", "watsonx-direct-adapter"],
            provider_data=WatsonxFlowArtifactProviderData(
                project_id=self.project_id,
                source_ref=str(flow_id),
            ),
        )

    def _build_config_payload(self, *, label: str) -> DeploymentConfig:
        return DeploymentConfig(
            name=self._mk_name(label),
            description="direct adapter config payload",
            environment_variables={},
        )

    def _build_flow_data_payload(self) -> dict[str, Any]:
        chat_input_node_id = f"ChatInput-{uuid4().hex[:8]}"
        chat_output_node_id = f"ChatOutput-{uuid4().hex[:8]}"
        return {
            "nodes": [
                {
                    "id": chat_input_node_id,
                    "type": "genericNode",
                    "position": {"x": 100, "y": 100},
                    "data": {
                        "type": "ChatInput",
                        "id": chat_input_node_id,
                        "node": {"template": {"_type": "CustomComponent"}},
                    },
                },
                {
                    "id": chat_output_node_id,
                    "type": "genericNode",
                    "position": {"x": 400, "y": 100},
                    "data": {
                        "type": "ChatOutput",
                        "id": chat_output_node_id,
                        "node": {"template": {"_type": "CustomComponent"}},
                    },
                },
            ],
            "edges": [],
            "viewport": {"x": 0, "y": 0, "zoom": 1},
        }

    async def _cleanup_resources(self) -> None:
        clients = await self.service._get_provider_clients(user_id=self.user_id, db=self.db)  # noqa: SLF001
        print("\nCleanup Resources")
        print("-" * 90)
        print(
            f"cleanup targets: deployments={len(self.created_deployment_ids)} "
            f"snapshots={len(self.created_snapshot_ids)} "
            f"configs={len(self.created_config_ids)}"
        )

        deleted_deployments = 0
        deleted_snapshots = 0
        deleted_configs = 0

        for deployment_id in sorted(self.created_deployment_ids):
            print(f"[cleanup] deleting deployment {deployment_id}...")
            with suppress(Exception):
                await self.service.delete(user_id=self.user_id, deployment_id=deployment_id, db=self.db)
                deleted_deployments += 1
                print(f"[cleanup] deleted deployment {deployment_id}")

        for snapshot_id in sorted(self.created_snapshot_ids):
            print(f"[cleanup] deleting snapshot {snapshot_id}...")
            try:
                await asyncio.to_thread(clients.tool.delete, snapshot_id)
                deleted_snapshots += 1
                print(f"[cleanup] deleted snapshot {snapshot_id}")
            except ClientAPIException as exc:
                if exc.response.status_code != HTTP_STATUS_NOT_FOUND:
                    print(f"[cleanup-warning] snapshot {snapshot_id}: {exc}")
                else:
                    print(f"[cleanup] snapshot {snapshot_id} already deleted (404)")

        for config_id in sorted(self.created_config_ids):
            print(f"[cleanup] deleting config {config_id}...")
            try:
                await asyncio.to_thread(clients.connections.delete, config_id)
                deleted_configs += 1
                print(f"[cleanup] deleted config {config_id}")
            except ClientAPIException as exc:
                if exc.response.status_code != HTTP_STATUS_NOT_FOUND:
                    print(f"[cleanup-warning] config {config_id}: {exc}")
                else:
                    print(f"[cleanup] config {config_id} already deleted (404)")

        print(
            f"cleanup completed: deployments_deleted={deleted_deployments} "
            f"snapshots_deleted={deleted_snapshots} "
            f"configs_deleted={deleted_configs}"
        )
        print("-" * 90)

    def _mk_name(self, prefix: str) -> str:
        raw = f"{prefix}_{self.run_suffix}_{uuid4().hex[:6]}"
        normalized = _INVALID_WXO_NAME_CHARS.sub("_", raw)
        if not normalized or not normalized[0].isalpha():
            normalized = f"n_{normalized}"
        return normalized

    def _scenario_group(self, name: str) -> str:
        prefix = name.split("_", 1)[0]
        if prefix in {"live", "upd", "cc", "fp"}:
            return prefix
        return "other"

    def _scenario_label(self, name: str) -> str:
        if "_" not in name:
            return name
        return name.split("_", 1)[1]

    def _format_expected_outcomes(self, outcomes: set[str]) -> str:
        ordered = sorted(outcomes, key=lambda item: (item != OUTCOME_SUCCESS, item))
        return ", ".join(ordered)

    def _print_result_row(self, result: ScenarioResult) -> None:
        verdict = "PASS" if result.ok else "FAIL"
        scenario_label = self._scenario_label(result.name)
        expected = self._format_expected_outcomes(result.expected_outcomes)
        print(f"{verdict:<5} | {scenario_label:<44} | got: {result.actual_outcome}")
        print(f"{'':5} | {'':44} | expected: {expected}")
        if not result.ok:
            wrapped_detail = textwrap.wrap(result.detail, width=108) or [result.detail]
            for index, line in enumerate(wrapped_detail):
                prefix = "detail: " if index == 0 else "        "
                print(f"{'':5} | {'':44} | {prefix}{line}")

    def _print_summary(self, results: list[ScenarioResult]) -> None:
        print("\nScenario Summary")
        print("-" * 90)
        total = len(results)
        passed = sum(1 for result in results if result.ok)
        failed = total - passed
        print(f"Total={total} Passed={passed} Failed={failed}")
        print("-" * 90)

        grouped_results: dict[str, list[ScenarioResult]] = {"live": [], "upd": [], "cc": [], "fp": [], "other": []}
        for result in results:
            grouped_results[self._scenario_group(result.name)].append(result)

        for group in ("live", "upd", "cc", "fp", "other"):
            group_items = grouped_results[group]
            if not group_items:
                continue
            print(f"\n[{group}] ({len(group_items)})")
            for result in group_items:
                self._print_result_row(result)
        print("-" * 90)


def _get_required_env(name: str) -> str:
    value = os.getenv(name, "").strip()
    if not value:
        msg = f"Environment variable '{name}' is required."
        raise RuntimeError(msg)
    return value


def _parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(description="Run direct Watsonx adapter matrix (live + failpoints).")
    parser.add_argument("--project-id", default=os.getenv("WXO_PROJECT_ID", "e2e-project"))
    parser.add_argument("--mode", choices=["live", "failpoint", "both"], default=os.getenv("WXO_E2E_MODE", "both"))
    parser.add_argument("--llm", default=os.getenv("WXO_DEFAULT_LLM", DEFAULT_WXO_LLM))
    parser.add_argument("--keep-resources", action="store_true")
    return parser.parse_args()


async def _main() -> int:
    load_dotenv()
    args = _parse_args()
    runner = WatsonxAdapterDirectE2E(
        provider_backend_url=_get_required_env("WXO_INSTANCE_URL"),
        provider_api_key=_get_required_env("WXO_API_KEY"),
        project_id=args.project_id,
        mode=args.mode,
        keep_resources=args.keep_resources,
        llm=args.llm,
    )
    return await runner.run()


if __name__ == "__main__":
    raise SystemExit(asyncio.run(_main()))