Update docker/docker-compose.yaml

.github/workflows/web-e2e.yml

@@ -63,6 +63,7 @@ jobs:
           E2E_ADMIN_PASSWORD: E2eAdmin12345
           E2E_FORCE_WEB_BUILD: "1"
           E2E_INIT_PASSWORD: E2eInit12345
+          E2E_START_AGENT_BACKEND: "1"
         run: vp run e2e:full
 
       - name: Preserve Chromium E2E report and logs
@@ -138,21 +139,6 @@ jobs:
             exit 1
           fi
 
-          teardown_external_runtime() {
-            local run_status=$?
-            trap - EXIT
-            if ! vp run e2e:middleware:down; then
-              echo "::error title=E2E teardown failed::External runtime middleware did not shut down cleanly."
-              if [[ "$run_status" -eq 0 ]]; then
-                run_status=1
-              fi
-            fi
-            exit "$run_status"
-          }
-
-          trap teardown_external_runtime EXIT
-          vp run e2e:middleware:up
-          vp run e2e:post-merge:prepare
           vp run e2e:post-merge
 
       - name: Upload Cucumber report

api/core/app/apps/agent_app/app_runner.py

@@ -634,7 +634,7 @@ class AgentAppRunner:
         agent_config_snapshot_id: str,
         agent_config_version_kind: Literal["snapshot", "draft", "build_draft"] = "snapshot",
         agent_soul: AgentSoulConfig,
-        home_snapshot_id: str | None,
+        home_snapshot_id: str,
         conversation_id: str,
         query: str,
         message_id: str,
@@ -736,7 +736,7 @@ class AgentAppRunner:
         dify_context: DifyRunContext,
         agent_id: str,
         agent_config_snapshot_id: str,
-        home_snapshot_id: str | None,
+        home_snapshot_id: str,
         conversation_id: str,
         session_scope_snapshot_id: str | None | _DefaultSessionScopeSnapshotId,
         agent_config_version_kind: AgentConfigVersionKind,

api/core/app/apps/agent_app/session_store.py

@@ -31,7 +31,7 @@ class AgentAppSessionScope:
     conversation_id: str
     agent_id: str
     agent_config_snapshot_id: str
-    home_snapshot_id: str | None
+    home_snapshot_id: str
     agent_config_version_kind: AgentConfigVersionKind = AgentConfigVersionKind.SNAPSHOT
     build_draft_id: str | None = None
 

api/core/workflow/nodes/agent_v2/session_store.py

@@ -125,7 +125,7 @@ class WorkflowAgentWorkspaceStore:
             )
 
     def load_or_create_node_execution_session(
-        self, scope: WorkflowAgentSessionScope, *, home_snapshot_id: str | None
+        self, scope: WorkflowAgentSessionScope, *, home_snapshot_id: str
     ) -> StoredWorkflowAgentSession:
         with session_factory.create_session() as session:
             execution = self._load_execution(session=session, scope=scope)

api/migrations/versions/2026_07_21_2251-2f39536b3feb_add_agent_home_snapshot_ledger.py

@@ -31,13 +31,13 @@ def upgrade():
         batch_op.create_index('agent_home_snapshot_tenant_agent_idx', ['tenant_id', 'agent_id'], unique=False)
 
     with op.batch_alter_table('agent_config_drafts', schema=None) as batch_op:
-        batch_op.add_column(sa.Column('home_snapshot_id', models.types.StringUUID(), nullable=True))
+        batch_op.add_column(sa.Column('home_snapshot_id', models.types.StringUUID(), nullable=False))
 
     with op.batch_alter_table('agent_config_snapshots', schema=None) as batch_op:
-        batch_op.add_column(sa.Column('home_snapshot_id', models.types.StringUUID(), nullable=True))
+        batch_op.add_column(sa.Column('home_snapshot_id', models.types.StringUUID(), nullable=False))
 
     with op.batch_alter_table('agent_runtime_sessions', schema=None) as batch_op:
-        batch_op.add_column(sa.Column('home_snapshot_id', models.types.StringUUID(), nullable=True))
+        batch_op.add_column(sa.Column('home_snapshot_id', models.types.StringUUID(), nullable=False))
         batch_op.drop_index(batch_op.f('agent_runtime_session_conversation_scope_unique'), postgresql_where='(conversation_id IS NOT NULL)')
         batch_op.create_index('agent_runtime_session_conversation_scope_unique', ['tenant_id', 'conversation_id', 'agent_id', 'agent_config_snapshot_id', 'home_snapshot_id'], unique=True, postgresql_where=sa.text('conversation_id IS NOT NULL'))
 

api/migrations/versions/2026_07_23_0203-f6e4c5686857_replace_agent_runtime_sessions_with_.py

@@ -24,7 +24,7 @@ def upgrade():
     sa.Column('app_id', models.types.StringUUID(), nullable=False),
     sa.Column('workspace_id', models.types.StringUUID(), nullable=False),
     sa.Column('agent_id', models.types.StringUUID(), nullable=False),
-    sa.Column('base_home_snapshot_id', models.types.StringUUID(), nullable=True),
+    sa.Column('base_home_snapshot_id', models.types.StringUUID(), nullable=False),
     sa.Column('agent_config_version_id', models.types.StringUUID(), nullable=False),
     sa.Column('agent_config_version_kind', sa.String(length=32), nullable=False),
     sa.Column('backend_binding_ref', sa.String(length=255), nullable=False),
@@ -127,7 +127,7 @@ def downgrade():
     sa.Column('updated_at', postgresql.TIMESTAMP(), server_default=sa.text('CURRENT_TIMESTAMP'), autoincrement=False, nullable=False),
     sa.Column('pending_form_id', sa.UUID(), autoincrement=False, nullable=True),
     sa.Column('pending_tool_call_id', sa.VARCHAR(length=255), autoincrement=False, nullable=True),
-    sa.Column('home_snapshot_id', sa.UUID(), autoincrement=False, nullable=True),
+    sa.Column('home_snapshot_id', sa.UUID(), autoincrement=False, nullable=False),
     sa.PrimaryKeyConstraint('id', name=op.f('agent_runtime_session_pkey'))
     )
     with op.batch_alter_table('agent_runtime_sessions', schema=None) as batch_op:

api/migrations/versions/2026_07_28_2331-e4708db55c1d_make_home_snapshot_references_nullable.py

@@ -1,49 +0,0 @@
-"""make home snapshot references nullable
-
-Revision ID: e4708db55c1d
-Revises: f6e4c5686857
-Create Date: 2026-07-28 23:31:26.284147
-
-The corrected preceding revisions make fresh upgrades nullable. This linear
-convergence revision intentionally repeats those alters so databases that
-already ran the old NOT NULL revisions are repaired too.
-
-"""
-from alembic import op
-import models as models
-
-
-# revision identifiers, used by Alembic.
-revision = 'e4708db55c1d'
-down_revision = 'f6e4c5686857'
-branch_labels = None
-depends_on = None
-
-
-def upgrade():
-    with op.batch_alter_table("agent_config_drafts", schema=None) as batch_op:
-        batch_op.alter_column(
-            "home_snapshot_id",
-            existing_type=models.types.StringUUID(),
-            nullable=True,
-        )
-
-    with op.batch_alter_table("agent_config_snapshots", schema=None) as batch_op:
-        batch_op.alter_column(
-            "home_snapshot_id",
-            existing_type=models.types.StringUUID(),
-            nullable=True,
-        )
-
-    with op.batch_alter_table("agent_workspace_bindings", schema=None) as batch_op:
-        batch_op.alter_column(
-            "base_home_snapshot_id",
-            existing_type=models.types.StringUUID(),
-            nullable=True,
-        )
-
-
-def downgrade():
-    # The corrected down revision already defines nullable columns, and rows
-    # created under this revision may contain NULL values.
-    pass

api/models/agent.py

@@ -307,7 +307,7 @@ class AgentConfigDraft(DefaultFieldsMixin, Base):
     account_id: Mapped[str | None] = mapped_column(StringUUID, nullable=True)
     draft_owner_key: Mapped[str] = mapped_column(String(255), nullable=False, default="")
     base_snapshot_id: Mapped[str | None] = mapped_column(StringUUID, nullable=True)
-    home_snapshot_id: Mapped[str | None] = mapped_column(StringUUID, nullable=True)
+    home_snapshot_id: Mapped[str] = mapped_column(StringUUID, nullable=False)
     agent_workspace_binding_id: Mapped[str | None] = mapped_column(StringUUID, nullable=True)
     config_snapshot: Mapped[Any] = mapped_column(JSONModelColumn(AgentSoulConfig), nullable=False)
     created_by: Mapped[str | None] = mapped_column(StringUUID, nullable=True)
@@ -344,7 +344,7 @@ class AgentConfigSnapshot(DefaultFieldsMixin, Base):
     agent_id: Mapped[str] = mapped_column(StringUUID, nullable=False)
     version: Mapped[int] = mapped_column(sa.Integer, nullable=False)
     config_snapshot: Mapped[Any] = mapped_column(JSONModelColumn(AgentSoulConfig), nullable=False)
-    home_snapshot_id: Mapped[str | None] = mapped_column(StringUUID, nullable=True)
+    home_snapshot_id: Mapped[str] = mapped_column(StringUUID, nullable=False)
     summary: Mapped[str | None] = mapped_column(LongText, nullable=True)
     version_note: Mapped[str | None] = mapped_column(LongText, nullable=True)
     created_by: Mapped[str | None] = mapped_column(StringUUID, nullable=True)
@@ -520,7 +520,7 @@ class AgentWorkspaceBinding(DefaultFieldsMixin, Base):
     app_id: Mapped[str] = mapped_column(StringUUID, nullable=False)
     workspace_id: Mapped[str] = mapped_column(StringUUID, nullable=False)
     agent_id: Mapped[str] = mapped_column(StringUUID, nullable=False)
-    base_home_snapshot_id: Mapped[str | None] = mapped_column(StringUUID, nullable=True)
+    base_home_snapshot_id: Mapped[str] = mapped_column(StringUUID, nullable=False)
     agent_config_version_id: Mapped[str] = mapped_column(StringUUID, nullable=False)
     agent_config_version_kind: Mapped[AgentConfigVersionKind] = mapped_column(
         EnumText(AgentConfigVersionKind, length=32), nullable=False

api/services/agent/composer_service.py

@@ -510,6 +510,11 @@ class AgentComposerService:
             )
             session.add(agent)
             session.flush()
+            home_snapshot = AgentHomeSnapshotService.create_initial(
+                session=session,
+                tenant_id=tenant_id,
+                agent_id=agent.id,
+            )
             initial_version = cls._create_config_version(
                 session=session,
                 tenant_id=tenant_id,
@@ -518,7 +523,7 @@ class AgentComposerService:
                 agent_soul=AgentSoulConfig(),
                 operation=AgentConfigRevisionOperation.CREATE_VERSION,
                 version_note=None,
-                home_snapshot_id=None,
+                home_snapshot_id=home_snapshot.id,
             )
             agent.active_config_snapshot_id = initial_version.id
             agent.active_config_has_model = False
@@ -596,7 +601,7 @@ class AgentComposerService:
         tenant_id: str,
         agent: Agent,
         agent_soul: AgentSoulConfig,
-        home_snapshot_id: str | None,
+        home_snapshot_id: str,
     ) -> bool:
         if not agent.active_config_snapshot_id:
             return False
@@ -1762,6 +1767,11 @@ class AgentComposerService:
         )
         session.add(agent)
         session.flush()
+        home_snapshot = AgentHomeSnapshotService.create_initial(
+            session=session,
+            tenant_id=tenant_id,
+            agent_id=agent.id,
+        )
         version = cls._create_config_version(
             session=session,
             tenant_id=tenant_id,
@@ -1770,7 +1780,7 @@ class AgentComposerService:
             agent_soul=agent_soul,
             operation=AgentConfigRevisionOperation.CREATE_VERSION,
             version_note=None,
-            home_snapshot_id=None,
+            home_snapshot_id=home_snapshot.id,
         )
         agent.active_config_snapshot_id = version.id
         agent.active_config_has_model = agent_soul_has_model(agent_soul)
@@ -1941,7 +1951,7 @@ class AgentComposerService:
         agent_soul: AgentSoulConfig,
         operation: AgentConfigRevisionOperation,
         version_note: str | None,
-        home_snapshot_id: str | None,
+        home_snapshot_id: str,
         previous_snapshot_id: str | None = None,
     ) -> AgentConfigSnapshot:
         next_version = (

api/services/agent/dsl_service.py

@@ -49,6 +49,7 @@ from services.agent.dsl_entities import (
     make_portable_agent_package,
     portable_ref,
 )
+from services.agent.home_snapshot_service import AgentHomeSnapshotService
 from services.agent.knowledge_datasets import get_tenant_knowledge_dataset_rows
 from services.agent.roster_service import AgentRosterService
 from services.entities.dsl_entities import DslImportWarning
@@ -568,12 +569,17 @@ class AgentDslService:
             )
             or 0
         ) + 1
+        home_snapshot = AgentHomeSnapshotService.create_initial(
+            session=self.session,
+            tenant_id=tenant_id,
+            agent_id=agent.id,
+        )
         snapshot = AgentConfigSnapshot(
             tenant_id=tenant_id,
             agent_id=agent.id,
             version=next_version,
             config_snapshot=soul,
-            home_snapshot_id=None,
+            home_snapshot_id=home_snapshot.id,
             created_by=account_id,
         )
         self.session.add(snapshot)

api/services/agent/home_snapshot_service.py

@@ -5,7 +5,7 @@ from __future__ import annotations
 import logging
 
 from dify_agent.client import Client, DifyAgentNotFoundError
-from dify_agent.protocol import CreateHomeSnapshotFromBindingRequest
+from dify_agent.protocol import CreateHomeSnapshotFromBindingRequest, InitializeHomeSnapshotRequest
 from sqlalchemy import select
 from sqlalchemy.orm import Session
 
@@ -36,6 +36,34 @@ class AgentHomeSnapshotUnavailableError(RuntimeError):
 class AgentHomeSnapshotService:
     """Create, retire, and collect Agent-owned immutable Home Snapshots."""
 
+    @classmethod
+    def create_initial(
+        cls,
+        *,
+        session: Session,
+        tenant_id: str,
+        agent_id: str,
+    ) -> AgentHomeSnapshot:
+        home_snapshot_id = str(uuidv7())
+        with cls._client() as client:
+            response = client.initialize_home_snapshot_sync(
+                InitializeHomeSnapshotRequest(
+                    tenant_id=tenant_id,
+                    agent_id=agent_id,
+                    home_snapshot_id=home_snapshot_id,
+                )
+            )
+        home_snapshot = AgentHomeSnapshot(
+            id=home_snapshot_id,
+            tenant_id=tenant_id,
+            agent_id=agent_id,
+            snapshot_ref=response.snapshot_ref,
+            status=AgentWorkingResourceStatus.ACTIVE,
+        )
+        session.add(home_snapshot)
+        session.flush()
+        return home_snapshot
+
     @classmethod
     def create_for_build_apply(
         cls,
@@ -183,9 +211,7 @@ class AgentHomeSnapshotService:
         return Client(base_url=base_url)
 
 
-def validate_home_snapshot_binding(*, session: Session, agent: Agent, home_snapshot_id: str | None) -> None:
-    if home_snapshot_id is None:
-        return
+def validate_home_snapshot_binding(*, session: Session, agent: Agent, home_snapshot_id: str) -> None:
     _require_owned_home_snapshot(session=session, agent=agent, home_snapshot_id=home_snapshot_id)
 
 

api/services/agent/roster_service.py

@@ -350,12 +350,17 @@ class AgentRosterService:
         self._session.add(agent)
         self._session.flush()
 
+        home_snapshot = AgentHomeSnapshotService.create_initial(
+            session=self._session,
+            tenant_id=tenant_id,
+            agent_id=agent.id,
+        )
         version = AgentConfigSnapshot(
             tenant_id=tenant_id,
             agent_id=agent.id,
             version=1,
             config_snapshot=payload.agent_soul,
-            home_snapshot_id=None,
+            home_snapshot_id=home_snapshot.id,
             version_note=payload.version_note,
             created_by=account_id,
         )
@@ -424,12 +429,17 @@ class AgentRosterService:
         self._session.add(agent)
         self._session.flush()
 
+        home_snapshot = AgentHomeSnapshotService.create_initial(
+            session=self._session,
+            tenant_id=tenant_id,
+            agent_id=agent.id,
+        )
         version = AgentConfigSnapshot(
             tenant_id=tenant_id,
             agent_id=agent.id,
             version=1,
             config_snapshot=soul,
-            home_snapshot_id=None,
+            home_snapshot_id=home_snapshot.id,
             created_by=account_id,
         )
         self._session.add(version)

api/services/agent/workspace_service.py

@@ -110,7 +110,7 @@ class AgentWorkspaceService:
         session: Session,
         scope: WorkspaceOwnerScope,
         agent_id: str,
-        base_home_snapshot_id: str | None,
+        base_home_snapshot_id: str,
         agent_config_version_id: str,
         agent_config_version_kind: AgentConfigVersionKind,
     ) -> AgentWorkspaceBinding:
@@ -123,19 +123,16 @@ class AgentWorkspaceService:
         fails before the backend returns success.
         """
 
-        home_snapshot_ref: str | None = None
-        if base_home_snapshot_id is not None:
-            home_snapshot = session.scalar(
-                select(AgentHomeSnapshot).where(
-                    AgentHomeSnapshot.id == base_home_snapshot_id,
-                    AgentHomeSnapshot.tenant_id == scope.tenant_id,
-                    AgentHomeSnapshot.agent_id == agent_id,
-                    AgentHomeSnapshot.status == AgentWorkingResourceStatus.ACTIVE,
-                )
+        home_snapshot = session.scalar(
+            select(AgentHomeSnapshot).where(
+                AgentHomeSnapshot.id == base_home_snapshot_id,
+                AgentHomeSnapshot.tenant_id == scope.tenant_id,
+                AgentHomeSnapshot.agent_id == agent_id,
+                AgentHomeSnapshot.status == AgentWorkingResourceStatus.ACTIVE,
             )
-            if home_snapshot is None:
-                raise AgentWorkspaceNotFoundError("base Home Snapshot is unavailable")
-            home_snapshot_ref = home_snapshot.snapshot_ref
+        )
+        if home_snapshot is None:
+            raise AgentWorkspaceNotFoundError("base Home Snapshot is unavailable")
         workspace = cls.resolve_active_workspace(session=session, scope=scope)
         workspace_id = workspace.id if workspace is not None else str(uuidv7())
         binding_id = str(uuidv7())
@@ -147,7 +144,7 @@ class AgentWorkspaceService:
                     binding_id=binding_id,
                     workspace_id=workspace_id,
                     existing_workspace_ref=workspace.backend_workspace_ref if workspace is not None else None,
-                    home_snapshot_ref=home_snapshot_ref,
+                    home_snapshot_ref=home_snapshot.snapshot_ref,
                 )
             )
         if workspace is not None and allocation.workspace_ref != workspace.backend_workspace_ref:
@@ -442,7 +439,7 @@ class AgentWorkspaceService:
     def validate_binding_generation(
         binding: AgentWorkspaceBinding,
         *,
-        base_home_snapshot_id: str | None,
+        base_home_snapshot_id: str,
         agent_config_version_id: str,
         agent_config_version_kind: AgentConfigVersionKind,
     ) -> None:

api/tests/test_containers_integration_tests/services/test_app_dsl_service.py

@@ -124,6 +124,7 @@ class TestAppDslService:
             patch("services.app_service.ModelManager.for_tenant") as mock_model_manager,
             patch("services.app_service.FeatureService") as mock_feature_service,
             patch("services.app_service.EnterpriseService") as mock_enterprise_service,
+            patch("services.agent.home_snapshot_service.AgentHomeSnapshotService._client") as mock_home_snapshot_client,
         ):
             mock_workflow_service.return_value.get_draft_workflow.return_value = None
             mock_workflow_service.return_value.sync_draft_workflow.return_value = MagicMock()
@@ -142,6 +143,10 @@ class TestAppDslService:
             mock_feature_service.get_system_features.return_value.webapp_auth.enabled = False
             mock_enterprise_service.WebAppAuth.update_app_access_mode.return_value = None
             mock_enterprise_service.WebAppAuth.cleanup_webapp.return_value = None
+            mock_home_snapshot_client.return_value.__enter__.return_value.initialize_home_snapshot_sync.side_effect = (
+                lambda request: SimpleNamespace(snapshot_ref=f"test:{request.home_snapshot_id}")
+            )
+
             yield {
                 "workflow_service": mock_workflow_service,
                 "dependencies_service": mock_dependencies_service,

api/tests/test_containers_integration_tests/services/test_app_service.py

@@ -1,4 +1,5 @@
 from datetime import datetime
+from types import SimpleNamespace
 from unittest.mock import create_autospec, patch
 
 import pytest
@@ -29,6 +30,7 @@ class TestAppService:
             patch("services.app_service.EnterpriseService") as mock_enterprise_service,
             patch("services.app_service.ModelManager.for_tenant") as mock_model_manager,
             patch("services.account_service.FeatureService") as mock_account_feature_service,
+            patch("services.agent.home_snapshot_service.AgentHomeSnapshotService._client") as mock_home_snapshot_client,
         ):
             # Setup default mock returns for app service
             mock_feature_service.get_system_features.return_value.webapp_auth.enabled = False
@@ -42,6 +44,10 @@ class TestAppService:
             mock_model_instance = mock_model_manager.return_value
             mock_model_instance.get_default_model_instance.return_value = None
             mock_model_instance.get_default_provider_model_name.return_value = ("openai", "gpt-3.5-turbo")
+            mock_home_snapshot_client.return_value.__enter__.return_value.initialize_home_snapshot_sync.side_effect = (
+                lambda request: SimpleNamespace(snapshot_ref=f"test:{request.home_snapshot_id}")
+            )
+
             yield {
                 "feature_service": mock_feature_service,
                 "enterprise_service": mock_enterprise_service,

api/tests/unit_tests/conftest.py

@@ -43,7 +43,7 @@ from models.account import Account, Tenant, TenantAccountJoin, TenantAccountRole
 from models.base import TypeBase
 
 
-def _patch_redis_clients_on_loaded_modules():
+def _patch_redis_clients_on_loaded_modules() -> None:
     """Ensure any module-level redis_client references point to the shared redis_mock."""
 
     import sys
@@ -51,10 +51,9 @@ def _patch_redis_clients_on_loaded_modules():
     for module in list(sys.modules.values()):
         if module is None:
             continue
-        if hasattr(module, "redis_client"):
-            module.redis_client = redis_mock
-        if hasattr(module, "_pubsub_redis_client"):
-            module.pubsub_redis_client = redis_mock
+        for client_attribute in ("redis_client", "_pubsub_redis_client"):
+            if hasattr(module, client_attribute):
+                setattr(module, client_attribute, redis_mock)
 
 
 @pytest.fixture
@@ -63,13 +62,13 @@ def app() -> Flask:
 
 
 @pytest.fixture(autouse=True)
-def _provide_app_context(app: Flask):
+def _provide_app_context(app: Flask) -> Iterator[None]:
     with app.app_context():
         yield
 
 
 @pytest.fixture(autouse=True)
-def _patch_redis_clients():
+def _patch_redis_clients() -> Iterator[None]:
     """Patch redis_client to MagicMock only for unit test executions."""
 
     with (
@@ -81,7 +80,7 @@ def _patch_redis_clients():
 
 
 @pytest.fixture(autouse=True)
-def reset_redis_mock():
+def reset_redis_mock() -> None:
     """reset the Redis mock before each test"""
     redis_mock.reset_mock()
     redis_mock.get.return_value = None
@@ -91,7 +90,7 @@ def reset_redis_mock():
     redis_mock.exists.return_value = False
     redis_mock.set.return_value = None
     redis_mock.expire.return_value = None
-    redis_mock.hgetall.return_value = {}
+    redis_mock.hgetall.return_value = dict[bytes, bytes]()
     redis_mock.hdel.return_value = None
     redis_mock.incr.return_value = 1
 
@@ -100,7 +99,7 @@ def reset_redis_mock():
 
 
 @pytest.fixture(autouse=True)
-def reset_secret_key():
+def reset_secret_key() -> Iterator[None]:
     """Ensure SECRET_KEY-dependent logic sees an empty config value by default."""
 
     from configs import dify_config
@@ -153,6 +152,18 @@ def _sqlite_session_factory(
     return factory
 
 
+@pytest.fixture
+def _unbound_session_factory(
+    _sqlite_session_factory: sessionmaker[Session],
+    monkeypatch: pytest.MonkeyPatch,
+) -> sessionmaker[Session]:
+    """Create one unbound factory and install it as the global test factory."""
+
+    factory = sessionmaker()
+    monkeypatch.setattr(session_factory_module, "_session_maker", factory)
+    return factory
+
+
 @pytest.fixture
 def sqlite_engine(_sqlite_engine: Engine) -> Engine:
     """Expose the pristine full-schema SQLite engine to tests."""
@@ -179,6 +190,25 @@ def sqlite_session(_sqlite_session_factory: sessionmaker[Session]) -> Iterator[S
         yield session
 
 
+@pytest.fixture
+def unbound_session_factory(_unbound_session_factory: sessionmaker[Session]) -> sessionmaker[Session]:
+    """Expose an unbound factory for paths that must not require persistence."""
+
+    return _unbound_session_factory
+
+
+@pytest.fixture
+def unbound_session(_unbound_session_factory: sessionmaker[Session]) -> Iterator[Session]:
+    """Yield an unbound Session for paths that must not require persistence.
+
+    Bind-requiring database access fails, while bind-free Session operations can
+    still succeed.
+    """
+
+    with _unbound_session_factory() as session:
+        yield session
+
+
 def persist_service_api_tenant_owner(session: Session, tenant: Tenant, owner: Account) -> TenantAccountJoin:
     """Persist the owner identity resolved by service-API app authentication.
 

api/tests/unit_tests/core/app/apps/agent_app/test_session_store.py

@@ -16,7 +16,6 @@ def _scope(
     *,
     kind: AgentConfigVersionKind = AgentConfigVersionKind.SNAPSHOT,
     build_draft_id: str | None = None,
-    home_snapshot_id: str | None = "home-1",
 ) -> AgentAppSessionScope:
     return AgentAppSessionScope(
         tenant_id="tenant-1",
@@ -24,7 +23,7 @@ def _scope(
         conversation_id="conversation-1",
         agent_id="agent-1",
         agent_config_snapshot_id="config-1",
-        home_snapshot_id=home_snapshot_id,
+        home_snapshot_id="home-1",
         agent_config_version_kind=kind,
         build_draft_id=build_draft_id,
     )
@@ -55,8 +54,7 @@ def test_scope_selects_conversation_or_build_draft_workspace_owner() -> None:
     assert build_owner.owner_id == "build-draft-1"
 
 
-@pytest.mark.parametrize("home_snapshot_id", ["home-1", None])
-def test_load_or_create_persists_new_binding_on_caller(monkeypatch, home_snapshot_id: str | None) -> None:
+def test_load_or_create_persists_new_binding_on_caller(monkeypatch) -> None:
     caller = SimpleNamespace(agent_workspace_binding_id=None)
     context = MagicMock()
     session = context.__enter__.return_value
@@ -66,14 +64,13 @@ def test_load_or_create_persists_new_binding_on_caller(monkeypatch, home_snapsho
     monkeypatch.setattr(store, "_load_caller", MagicMock(return_value=caller))
     monkeypatch.setattr(AgentWorkspaceService, "create_binding", create)
 
-    stored = store.load_or_create(_scope(home_snapshot_id=home_snapshot_id))
+    stored = store.load_or_create(_scope())
 
     assert stored.binding_id == "binding-1"
     assert stored.workspace_id == "workspace-1"
     assert stored.backend_binding_ref == "backend-binding-1"
     assert caller.agent_workspace_binding_id == "binding-1"
     assert create.call_args.kwargs["session"] is session
-    assert create.call_args.kwargs["base_home_snapshot_id"] == home_snapshot_id
     session.commit.assert_called_once_with()
 
 

api/tests/unit_tests/core/workflow/nodes/agent_v2/test_session_store.py

@@ -155,8 +155,7 @@ def test_load_existing_scope_rejects_unavailable_persisted_binding(monkeypatch:
         )
 
 
-@pytest.mark.parametrize("home_snapshot_id", ["home-1", None])
-def test_load_or_create_persists_binding_on_node_execution(monkeypatch, home_snapshot_id: str | None) -> None:
+def test_load_or_create_persists_binding_on_node_execution(monkeypatch) -> None:
     execution = WorkflowNodeExecutionModel(
         agent_workspace_binding_id=None,
         process_data=json.dumps({"existing": "value"}),
@@ -172,7 +171,7 @@ def test_load_or_create_persists_binding_on_node_execution(monkeypatch, home_sna
     monkeypatch.setattr(store, "_load_execution", MagicMock(return_value=execution))
     monkeypatch.setattr(AgentWorkspaceService, "create_binding", create)
 
-    stored = store.load_or_create_node_execution_session(_scope(), home_snapshot_id=home_snapshot_id)
+    stored = store.load_or_create_node_execution_session(_scope(), home_snapshot_id="home-1")
 
     assert stored.binding_id == "binding-1"
     assert stored.workspace_id == "workspace-1"
@@ -184,7 +183,6 @@ def test_load_or_create_persists_binding_on_node_execution(monkeypatch, home_sna
     }
     assert "agent_workspace_binding_id" not in execution.process_data_dict
     assert create.call_args.kwargs["session"] is session
-    assert create.call_args.kwargs["base_home_snapshot_id"] == home_snapshot_id
     session.commit.assert_called_once_with()
 
     get_active = MagicMock(return_value=_binding())

api/tests/unit_tests/events/test_update_provider_when_message_created.py

@@ -11,7 +11,8 @@ from sqlalchemy.orm import Session, sessionmaker
 from core.app.entities.app_invoke_entities import ChatAppGenerateEntity
 from core.entities.provider_entities import ProviderQuotaType, QuotaUnit
 from events.event_handlers import update_provider_when_message_created
-from models import TenantCreditPool
+from models import Message, TenantCreditPool
+from models.enums import ProviderQuotaType as ModelProviderQuotaType
 from models.provider import ProviderType
 
 
@@ -30,7 +31,7 @@ def test_message_created_trial_credit_accounting_does_not_raise_when_balance_is_
     pool_id = str(uuid4())
     pool = TenantCreditPool(
         tenant_id=tenant_id,
-        pool_type=ProviderQuotaType.TRIAL,
+        pool_type=ModelProviderQuotaType.TRIAL,
         quota_limit=10,
         quota_used=9,
     )
@@ -61,7 +62,7 @@ def test_message_created_trial_credit_accounting_does_not_raise_when_balance_is_
             ),
         ),
     )
-    message = SimpleNamespace(message_tokens=2, answer_tokens=1)
+    message = Message(message_tokens=2, answer_tokens=1)
 
     with (
         patch.object(update_provider_when_message_created, "_execute_provider_updates"),
@@ -102,7 +103,7 @@ def test_message_created_paid_credit_accounting_uses_paid_pool() -> None:
             ),
         ),
     )
-    message = SimpleNamespace(message_tokens=2, answer_tokens=1)
+    message = Message(message_tokens=2, answer_tokens=1)
 
     with (
         patch.object(update_provider_when_message_created, "_deduct_credit_pool_quota_capped") as mock_deduct,

api/tests/unit_tests/migrations/test_agent_workspace_migration.py

@@ -42,7 +42,7 @@ def _create_pre_upgrade_schema(engine: sa.Engine) -> None:
         sa.Column("binding_id", sa.String(36)),
         sa.Column("agent_id", sa.String(36), nullable=False),
         sa.Column("agent_config_snapshot_id", sa.String(36)),
-        sa.Column("home_snapshot_id", sa.String(36), nullable=True),
+        sa.Column("home_snapshot_id", sa.String(36), nullable=False),
         sa.Column("backend_run_id", sa.String(255)),
         sa.Column("status", sa.String(32), nullable=False),
     )
@@ -110,10 +110,7 @@ def test_upgrade_replaces_runtime_sessions_with_workspace_schema() -> None:
 
     inspector = sa.inspect(engine)
     assert "agent_runtime_sessions" not in inspector.get_table_names()
-    binding_column_definitions = {
-        column["name"]: column for column in inspector.get_columns("agent_workspace_bindings")
-    }
-    binding_columns = set(binding_column_definitions)
+    binding_columns = {column["name"] for column in inspector.get_columns("agent_workspace_bindings")}
     assert {
         "workspace_id",
         "agent_id",
@@ -128,7 +125,6 @@ def test_upgrade_replaces_runtime_sessions_with_workspace_schema() -> None:
         "pending_tool_call_id",
     }.issubset(binding_columns)
     assert "active_guard" not in binding_columns
-    assert binding_column_definitions["base_home_snapshot_id"]["nullable"] is True
     binding_indexes = {index["name"] for index in inspector.get_indexes("agent_workspace_bindings")}
     assert "agent_workspace_binding_agent_active_unique" not in binding_indexes
     workspace_columns = {column["name"] for column in inspector.get_columns("agent_workspaces")}

api/tests/unit_tests/migrations/test_nullable_home_snapshot_migration.py

@@ -1,181 +0,0 @@
-from __future__ import annotations
-
-import importlib.util
-from collections.abc import Callable
-from pathlib import Path
-from typing import Protocol, cast
-
-import sqlalchemy as sa
-from alembic.migration import MigrationContext
-from alembic.operations import Operations
-
-_VERSIONS_DIR = Path(__file__).resolve().parents[3] / "migrations/versions"
-_MIGRATION_PATHS = (
-    _VERSIONS_DIR / "2026_07_21_2251-2f39536b3feb_add_agent_home_snapshot_ledger.py",
-    _VERSIONS_DIR / "2026_07_23_0203-f6e4c5686857_replace_agent_runtime_sessions_with_.py",
-    _VERSIONS_DIR / "2026_07_28_2331-e4708db55c1d_make_home_snapshot_references_nullable.py",
-)
-
-
-class _MigrationModule(Protocol):
-    op: Operations
-    upgrade: Callable[[], None]
-
-
-def _load_migration(path: Path) -> _MigrationModule:
-    spec = importlib.util.spec_from_file_location(path.stem, path)
-    if spec is None or spec.loader is None:
-        raise RuntimeError(f"failed to load migration {path.name}")
-    module = importlib.util.module_from_spec(spec)
-    spec.loader.exec_module(module)
-    return cast(_MigrationModule, module)
-
-
-def _run_upgrade(module: _MigrationModule, engine: sa.Engine) -> None:
-    with engine.begin() as connection:
-        original_op = module.op
-        module.op = Operations(MigrationContext.configure(connection))
-        try:
-            module.upgrade()
-        finally:
-            module.op = original_op
-
-
-def _create_pre_home_snapshot_schema(engine: sa.Engine) -> None:
-    metadata = sa.MetaData()
-    drafts = sa.Table("agent_config_drafts", metadata, sa.Column("id", sa.String(36), primary_key=True))
-    snapshots = sa.Table("agent_config_snapshots", metadata, sa.Column("id", sa.String(36), primary_key=True))
-    conversations = sa.Table("conversations", metadata, sa.Column("id", sa.String(36), primary_key=True))
-    executions = sa.Table("workflow_node_executions", metadata, sa.Column("id", sa.String(36), primary_key=True))
-    runtime_sessions = sa.Table(
-        "agent_runtime_sessions",
-        metadata,
-        sa.Column("id", sa.String(36), primary_key=True),
-        sa.Column("tenant_id", sa.String(36), nullable=False),
-        sa.Column("conversation_id", sa.String(36)),
-        sa.Column("workflow_run_id", sa.String(36)),
-        sa.Column("node_id", sa.String(255)),
-        sa.Column("binding_id", sa.String(36)),
-        sa.Column("agent_id", sa.String(36), nullable=False),
-        sa.Column("agent_config_snapshot_id", sa.String(36)),
-        sa.Column("backend_run_id", sa.String(255)),
-        sa.Column("status", sa.String(32), nullable=False),
-    )
-    sa.Index("agent_runtime_session_backend_run_idx", runtime_sessions.c.backend_run_id)
-    sa.Index(
-        "agent_runtime_session_conversation_lookup_idx",
-        runtime_sessions.c.tenant_id,
-        runtime_sessions.c.conversation_id,
-        runtime_sessions.c.status,
-    )
-    sa.Index(
-        "agent_runtime_session_conversation_scope_unique",
-        runtime_sessions.c.tenant_id,
-        runtime_sessions.c.conversation_id,
-        runtime_sessions.c.agent_id,
-        runtime_sessions.c.agent_config_snapshot_id,
-        unique=True,
-    )
-    sa.Index(
-        "agent_runtime_session_workflow_lookup_idx",
-        runtime_sessions.c.tenant_id,
-        runtime_sessions.c.workflow_run_id,
-        runtime_sessions.c.node_id,
-        runtime_sessions.c.status,
-    )
-    sa.Index(
-        "agent_runtime_session_workflow_scope_unique",
-        runtime_sessions.c.tenant_id,
-        runtime_sessions.c.workflow_run_id,
-        runtime_sessions.c.node_id,
-        runtime_sessions.c.binding_id,
-        runtime_sessions.c.agent_id,
-        unique=True,
-    )
-    metadata.create_all(engine)
-    with engine.begin() as connection:
-        connection.execute(drafts.insert().values(id="draft-1"))
-        connection.execute(snapshots.insert().values(id="snapshot-1"))
-        connection.execute(conversations.insert().values(id="conversation-1"))
-        connection.execute(executions.insert().values(id="execution-1"))
-        connection.execute(
-            runtime_sessions.insert().values(
-                id="runtime-1",
-                tenant_id="tenant-1",
-                conversation_id="conversation-1",
-                agent_id="agent-1",
-                agent_config_snapshot_id="snapshot-1",
-                status="active",
-            )
-        )
-
-
-def _create_old_f6_schema(engine: sa.Engine) -> None:
-    metadata = sa.MetaData()
-    drafts = sa.Table(
-        "agent_config_drafts",
-        metadata,
-        sa.Column("id", sa.String(36), primary_key=True),
-        sa.Column("home_snapshot_id", sa.String(36), nullable=False),
-    )
-    snapshots = sa.Table(
-        "agent_config_snapshots",
-        metadata,
-        sa.Column("id", sa.String(36), primary_key=True),
-        sa.Column("home_snapshot_id", sa.String(36), nullable=False),
-    )
-    bindings = sa.Table(
-        "agent_workspace_bindings",
-        metadata,
-        sa.Column("id", sa.String(36), primary_key=True),
-        sa.Column("base_home_snapshot_id", sa.String(36), nullable=False),
-    )
-    metadata.create_all(engine)
-    with engine.begin() as connection:
-        connection.execute(drafts.insert().values(id="draft-1", home_snapshot_id="home-draft"))
-        connection.execute(snapshots.insert().values(id="snapshot-1", home_snapshot_id="home-snapshot"))
-        connection.execute(bindings.insert().values(id="binding-1", base_home_snapshot_id="home-binding"))
-
-
-def test_historical_rows_upgrade_through_nullable_home_snapshot_chain() -> None:
-    engine = sa.create_engine("sqlite:///:memory:")
-    _create_pre_home_snapshot_schema(engine)
-
-    for path in _MIGRATION_PATHS:
-        _run_upgrade(_load_migration(path), engine)
-
-    inspector = sa.inspect(engine)
-    for table_name, column_name in (
-        ("agent_config_drafts", "home_snapshot_id"),
-        ("agent_config_snapshots", "home_snapshot_id"),
-        ("agent_workspace_bindings", "base_home_snapshot_id"),
-    ):
-        columns = {column["name"]: column for column in inspector.get_columns(table_name)}
-        assert columns[column_name]["nullable"] is True
-
-    with engine.connect() as connection:
-        assert connection.scalar(sa.text("SELECT home_snapshot_id FROM agent_config_drafts")) is None
-        assert connection.scalar(sa.text("SELECT home_snapshot_id FROM agent_config_snapshots")) is None
-
-
-def test_old_f6_schema_converges_to_nullable_without_changing_data() -> None:
-    engine = sa.create_engine("sqlite:///:memory:")
-    _create_old_f6_schema(engine)
-
-    _run_upgrade(_load_migration(_MIGRATION_PATHS[-1]), engine)
-
-    inspector = sa.inspect(engine)
-    for table_name, column_name in (
-        ("agent_config_drafts", "home_snapshot_id"),
-        ("agent_config_snapshots", "home_snapshot_id"),
-        ("agent_workspace_bindings", "base_home_snapshot_id"),
-    ):
-        columns = {column["name"]: column for column in inspector.get_columns(table_name)}
-        assert columns[column_name]["nullable"] is True
-
-    with engine.connect() as connection:
-        assert connection.scalar(sa.text("SELECT home_snapshot_id FROM agent_config_drafts")) == "home-draft"
-        assert connection.scalar(sa.text("SELECT home_snapshot_id FROM agent_config_snapshots")) == "home-snapshot"
-        assert (
-            connection.scalar(sa.text("SELECT base_home_snapshot_id FROM agent_workspace_bindings")) == "home-binding"
-        )

api/tests/unit_tests/pyrefly.toml

@@ -28,7 +28,6 @@ project-excludes = [
   "configs/test_dify_config.py",
   "configs/test_env_consistency.py",
   "configs/test_nacos_http_client.py",
-  "conftest.py",
   "controllers/common/test_agent_app_parameters.py",
   "controllers/common/test_app_access.py",
   "controllers/common/test_errors.py",
@@ -652,7 +651,6 @@ project-excludes = [
   "events/event_handlers/test_delete_tool_parameters_cache_when_sync_draft_workflow.py",
   "events/test_app_event_signals.py",
   "events/test_events_package_compat.py",
-  "events/test_update_provider_when_message_created.py",
   "extensions/logstore/repositories/test_logstore_api_workflow_node_execution_repository.py",
   "extensions/logstore/test_sql_escape.py",
   "extensions/otel/conftest.py",
@@ -863,7 +861,6 @@ project-excludes = [
   "services/test_file_service.py",
   "services/test_human_input_delivery_test_service.py",
   "services/test_human_input_file_upload_service.py",
-  "services/test_human_input_service.py",
   "services/test_knowledge_retrieval_inner_service.py",
   "services/test_knowledge_service.py",
   "services/test_message_service.py",

api/tests/unit_tests/services/agent/test_agent_dsl_service.py

@@ -684,10 +684,12 @@ def test_resolve_package_soul_preserves_existing_and_marks_missing_knowledge(mon
     }
 
 
-def test_create_snapshot_increments_version_and_records_revision() -> None:
+def test_create_snapshot_increments_version_and_records_revision(monkeypatch: pytest.MonkeyPatch) -> None:
     session = Mock()
     session.scalar.return_value = 2
     service = AgentDslService(session)
+    create_initial = Mock(return_value=SimpleNamespace(id="home-3"))
+    monkeypatch.setattr("services.agent.dsl_service.AgentHomeSnapshotService.create_initial", create_initial)
 
     snapshot = service._create_snapshot(
         tenant_id="tenant-1",
@@ -698,7 +700,12 @@ def test_create_snapshot_increments_version_and_records_revision() -> None:
     )
 
     assert snapshot.version == 3
-    assert snapshot.home_snapshot_id is None
+    assert snapshot.home_snapshot_id == "home-3"
+    create_initial.assert_called_once_with(
+        session=session,
+        tenant_id="tenant-1",
+        agent_id="agent-1",
+    )
     assert isinstance(session.add.call_args_list[0].args[0], AgentConfigSnapshot)
     revision = session.add.call_args_list[1].args[0]
     assert isinstance(revision, AgentConfigRevision)

api/tests/unit_tests/services/agent/test_agent_services.py

@@ -129,6 +129,15 @@ class FakeSession:
         self.rollbacks += 1
 
 
+@pytest.fixture
+def _stub_home_snapshot_backend(monkeypatch: pytest.MonkeyPatch):
+    create_initial = MagicMock(return_value=SimpleNamespace(id="home-initial", snapshot_ref="backend-home-initial"))
+    create_for_build_apply = MagicMock(return_value=SimpleNamespace(id="home-build", snapshot_ref="backend-home-build"))
+    monkeypatch.setattr(AgentHomeSnapshotService, "create_initial", create_initial)
+    monkeypatch.setattr(AgentHomeSnapshotService, "create_for_build_apply", create_for_build_apply)
+    return SimpleNamespace(create_initial=create_initial, create_for_build_apply=create_for_build_apply)
+
+
 def _agent_soul_with_model() -> AgentSoulConfig:
     return AgentSoulConfig.model_validate(
         {
@@ -603,6 +612,7 @@ def test_publish_save_strategies_run_publish_validation(strategy: ComposerSaveSt
         composer_service._validate_composer_payload_for_strategy(_duplicate_env_secret_payload(strategy))
 
 
+@pytest.mark.usefixtures("_stub_home_snapshot_backend")
 def test_save_agent_app_composer_creates_agent_when_missing(monkeypatch: pytest.MonkeyPatch):
     fake_session = FakeSession(scalar=[None])
     saved_draft = SimpleNamespace(
@@ -636,7 +646,7 @@ def test_save_agent_app_composer_creates_agent_when_missing(monkeypatch: pytest.
     assert result == {"loaded": True}
     assert fake_session.added[0].name == "Analyst"
     assert fake_session.added[0].active_config_snapshot_id == fake_session.added[1].id
-    assert fake_session.added[1].home_snapshot_id is None
+    assert fake_session.added[1].home_snapshot_id == "home-initial"
     assert fake_session.added[0].active_config_is_published is False
     assert fake_session.flushes >= 1
 
@@ -859,7 +869,7 @@ def test_publish_agent_app_draft_creates_published_snapshot(monkeypatch: pytest.
         draft_type=AgentConfigDraftType.DRAFT,
         draft_owner_key="",
         base_snapshot_id="version-1",
-        home_snapshot_id=None,
+        home_snapshot_id="home-1",
         config_snapshot=_agent_soul_with_model(),
     )
     version = SimpleNamespace(id="version-2")
@@ -895,7 +905,7 @@ def test_publish_agent_app_draft_creates_published_snapshot(monkeypatch: pytest.
     assert result["draft"]["base_snapshot_id"] == "version-2"
     assert created["operation"] == AgentConfigRevisionOperation.PUBLISH_DRAFT
     assert created["previous_snapshot_id"] == "version-1"
-    assert created["home_snapshot_id"] is None
+    assert created["home_snapshot_id"] == "home-1"
     assert calls == ["validate_home", "create_version"]
     assert agent.active_config_snapshot_id == "version-2"
     assert agent.active_config_has_model is True
@@ -929,6 +939,7 @@ def test_repeated_publish_reuses_normal_draft_home_without_creating_resources(
     session = FakeSession(scalar=[agent, draft, agent, draft])
     published_homes: list[str] = []
     versions = iter([SimpleNamespace(id="version-2"), SimpleNamespace(id="version-3")])
+    create_initial = MagicMock()
     create_from_build = MagicMock()
     monkeypatch.setattr(composer_service.ComposerConfigValidator, "validate_publish_payload", lambda _payload: None)
     monkeypatch.setattr(AgentComposerService, "validate_knowledge_datasets", lambda **_kwargs: None)
@@ -940,6 +951,7 @@ def test_repeated_publish_reuses_normal_draft_home_without_creating_resources(
     )
     monkeypatch.setattr(AgentComposerService, "_serialize_version", lambda version: {"id": version.id})
     monkeypatch.setattr(AgentComposerService, "_serialize_draft", lambda value: {"id": value.id})
+    monkeypatch.setattr(AgentHomeSnapshotService, "create_initial", create_initial)
     monkeypatch.setattr(AgentHomeSnapshotService, "create_for_build_apply", create_from_build)
 
     first = AgentComposerService.publish_agent_app_draft(
@@ -959,6 +971,7 @@ def test_repeated_publish_reuses_normal_draft_home_without_creating_resources(
     assert second["active_config_snapshot_id"] == "version-3"
     assert published_homes == ["home-1", "home-1"]
     assert draft.home_snapshot_id == "home-1"
+    create_initial.assert_not_called()
     create_from_build.assert_not_called()
 
 
@@ -1220,7 +1233,7 @@ def test_build_apply_checkpoints_binding_updates_normal_draft_then_collects(monk
         draft_type=AgentConfigDraftType.DEBUG_BUILD,
         account_id="account-1",
         draft_owner_key="account-1",
-        home_snapshot_id=None,
+        home_snapshot_id="home-old",
         agent_workspace_binding_id="binding-1",
         config_snapshot=AgentSoulConfig(),
     )
@@ -1230,14 +1243,14 @@ def test_build_apply_checkpoints_binding_updates_normal_draft_then_collects(monk
         agent_id="agent-1",
         draft_type=AgentConfigDraftType.DRAFT,
         draft_owner_key="",
-        home_snapshot_id=None,
+        home_snapshot_id="home-old",
         config_snapshot=AgentSoulConfig(),
     )
     source_binding = SimpleNamespace(
         id="binding-1",
         backend_binding_ref="backend-binding-1",
         agent_id="agent-1",
-        base_home_snapshot_id=None,
+        base_home_snapshot_id="home-old",
         agent_config_version_id="build-1",
         agent_config_version_kind=AgentConfigVersionKind.BUILD_DRAFT,
     )
@@ -1749,8 +1762,10 @@ def test_build_draft_save_and_discard_do_not_manage_home_resources(monkeypatch:
         home_snapshot_id="home-existing",
         config_snapshot=AgentSoulConfig(),
     )
+    create_initial = MagicMock()
     create_from_build = MagicMock()
     delete_home = MagicMock()
+    monkeypatch.setattr(AgentHomeSnapshotService, "create_initial", create_initial)
     monkeypatch.setattr(AgentHomeSnapshotService, "create_for_build_apply", create_from_build)
     monkeypatch.setattr(AgentHomeSnapshotService, "delete", delete_home)
     monkeypatch.setattr(AgentComposerService, "_require_agent", lambda **_kwargs: agent)
@@ -1776,6 +1791,7 @@ def test_build_draft_save_and_discard_do_not_manage_home_resources(monkeypatch:
         account_id="account-1",
     )
 
+    create_initial.assert_not_called()
     create_from_build.assert_not_called()
     delete_home.assert_not_called()
 
@@ -3152,6 +3168,7 @@ def test_composer_create_agents_syncs_active_config_has_model(
     session = fake_session
     created_apps = []
     hidden_backing_apps = []
+    create_initial = MagicMock(return_value=SimpleNamespace(id="home-initial", snapshot_ref="backend-home-initial"))
     backing_agent = Agent(
         id="roster-agent-1",
         tenant_id="tenant-1",
@@ -3182,6 +3199,7 @@ def test_composer_create_agents_syncs_active_config_has_model(
 
     monkeypatch.setattr(composer_service, "AppService", FakeAppService)
     monkeypatch.setattr(composer_service, "AgentRosterService", FakeAgentRosterService)
+    monkeypatch.setattr(AgentHomeSnapshotService, "create_initial", create_initial)
     monkeypatch.setattr(AgentComposerService, "_require_account", lambda **kwargs: SimpleNamespace(id="account-1"))
     monkeypatch.setattr(
         AgentComposerService,
@@ -3190,11 +3208,14 @@ def test_composer_create_agents_syncs_active_config_has_model(
             id="empty-version-1",
             tenant_id="tenant-1",
             agent_id="roster-agent-1",
-            home_snapshot_id=None,
+            home_snapshot_id="home-roster-initial",
         ),
     )
-    create_config_version = MagicMock(return_value=SimpleNamespace(id="version-with-model"))
-    monkeypatch.setattr(AgentComposerService, "_create_config_version", create_config_version)
+    monkeypatch.setattr(
+        AgentComposerService,
+        "_create_config_version",
+        lambda **kwargs: SimpleNamespace(id="version-with-model"),
+    )
 
     workflow_agent = AgentComposerService._create_workflow_only_agent(
         session=session,
@@ -3218,8 +3239,11 @@ def test_composer_create_agents_syncs_active_config_has_model(
     assert workflow_agent.active_config_snapshot_id == "version-with-model"
     assert workflow_agent.active_config_has_model is True
     assert workflow_agent.backing_app_id == "hidden-app-1"
-    assert create_config_version.call_count == 2
-    assert all(call.kwargs["home_snapshot_id"] is None for call in create_config_version.call_args_list)
+    create_initial.assert_called_once_with(
+        session=session,
+        tenant_id="tenant-1",
+        agent_id=workflow_agent.id,
+    )
     assert hidden_backing_apps[0]["name"] == "Workflow Agent node-1"
     assert roster_agent.active_config_snapshot_id == "version-with-model"
     assert roster_agent.active_config_has_model is True
@@ -3938,6 +3962,13 @@ def test_roster_create_detail_and_lookup_helpers(monkeypatch: pytest.MonkeyPatch
         scalars=[[AgentConfigSnapshot(id="version-1", agent_id="agent-1", version=1)]],
     )
     service = AgentRosterService(fake_session)
+    create_initial = MagicMock(
+        side_effect=[
+            SimpleNamespace(id="home-roster", snapshot_ref="backend-home-roster"),
+            SimpleNamespace(id="home-backing", snapshot_ref="backend-home-backing"),
+        ]
+    )
+    monkeypatch.setattr(AgentHomeSnapshotService, "create_initial", create_initial)
     monkeypatch.setattr(
         AgentRosterService,
         "_get_or_create_agent_app_debug_conversation",
@@ -3975,9 +4006,10 @@ def test_roster_create_detail_and_lookup_helpers(monkeypatch: pytest.MonkeyPatch
     assert created.role == "Research assistant"
     assert created.source == AgentSource.ROSTER
     assert created.active_config_snapshot_id is not None
+    assert create_initial.call_count == 2
     assert [
         snapshot.home_snapshot_id for snapshot in fake_session.added if isinstance(snapshot, AgentConfigSnapshot)
-    ] == [None, None]
+    ] == ["home-roster", "home-backing"]
     assert created.active_config_has_model is False
     assert backing_agent.role == "Support agent"
     assert backing_agent.active_config_snapshot_id is not None
@@ -4521,6 +4553,9 @@ class TestAgentAppBackingAgent:
     def test_create_backing_agent_for_app_links_app_and_seeds_default_soul(self, monkeypatch: pytest.MonkeyPatch):
         session = FakeSession()
         service = AgentRosterService(session)
+        create_initial = MagicMock(return_value=SimpleNamespace(id="home-1", snapshot_ref="backend-home-1"))
+        monkeypatch.setattr(AgentHomeSnapshotService, "create_initial", create_initial)
+
         agent = service.create_backing_agent_for_app(
             tenant_id="tenant-1",
             account_id="account-1",
@@ -4542,7 +4577,12 @@ class TestAgentAppBackingAgent:
         snapshots = [a for a in session.added if isinstance(a, AgentConfigSnapshot)]
         assert len(snapshots) == 1
         assert snapshots[0].version == 1
-        assert snapshots[0].home_snapshot_id is None
+        assert snapshots[0].home_snapshot_id == "home-1"
+        create_initial.assert_called_once_with(
+            session=session,
+            tenant_id="tenant-1",
+            agent_id=agent.id,
+        )
         assert agent.active_config_snapshot_id == snapshots[0].id
         revisions = [
             a for a in session.added if getattr(a, "operation", None) == AgentConfigRevisionOperation.CREATE_VERSION
@@ -4926,6 +4966,7 @@ class TestAgentAppBackingAgent:
             scalars=[[]],
         )
         captured: dict[str, object] = {}
+        create_initial = MagicMock(return_value=SimpleNamespace(id="home-target", snapshot_ref="backend-home-target"))
 
         class FakeAppService:
             def create_app(self, tenant_id: str, params, account: object, *, session) -> object:
@@ -4951,6 +4992,7 @@ class TestAgentAppBackingAgent:
                 return target_app
 
         monkeypatch.setattr(roster_service, "AppService", FakeAppService)
+        monkeypatch.setattr(AgentHomeSnapshotService, "create_initial", create_initial)
         monkeypatch.setattr(
             AgentRosterService,
             "_get_or_create_agent_app_debug_conversation",
@@ -4985,11 +5027,16 @@ class TestAgentAppBackingAgent:
         target_version = captured["target_version"]
         assert target_version.config_snapshot.model.model == "gpt-4o"
         assert source_version.home_snapshot_id == "home-source"
-        assert target_version.home_snapshot_id is None
+        assert target_version.home_snapshot_id == "home-target"
         assert target_version.summary == "configured"
         assert target_version.version_note == "v1"
         assert target_agent.active_config_has_model is True
         assert target_agent.updated_by == "account-1"
+        create_initial.assert_called_once_with(
+            session=session,
+            tenant_id="tenant-1",
+            agent_id=target_agent.id,
+        )
         assert session.commits == 1
 
     def test_duplicate_agent_app_inherits_webapp_access_mode(self, monkeypatch: pytest.MonkeyPatch):

api/tests/unit_tests/services/agent/test_home_snapshot_service.py

@@ -6,7 +6,6 @@ import pytest
 from sqlalchemy.orm import Session
 
 from models.agent import (
-    Agent,
     AgentConfigDraft,
     AgentConfigDraftType,
     AgentConfigSnapshot,
@@ -15,11 +14,11 @@ from models.agent import (
 )
 from models.agent_config_entities import AgentSoulConfig
 from services.agent.errors import AgentBuildSandboxNotFoundError
-from services.agent.home_snapshot_service import AgentHomeSnapshotService, validate_home_snapshot_binding
+from services.agent.home_snapshot_service import AgentHomeSnapshotService
 from services.agent.workspace_service import AgentWorkspaceService
 
 
-def _build_draft(*, home_snapshot_id: str | None = "home-old") -> AgentConfigDraft:
+def _build_draft() -> AgentConfigDraft:
     return AgentConfigDraft(
         id="build-1",
         tenant_id="tenant-1",
@@ -27,7 +26,7 @@ def _build_draft(*, home_snapshot_id: str | None = "home-old") -> AgentConfigDra
         draft_type=AgentConfigDraftType.DEBUG_BUILD,
         account_id="account-1",
         draft_owner_key="account-1",
-        home_snapshot_id=home_snapshot_id,
+        home_snapshot_id="home-old",
         agent_workspace_binding_id="binding-1",
         config_snapshot=AgentSoulConfig(),
     )
@@ -35,19 +34,42 @@ def _build_draft(*, home_snapshot_id: str | None = "home-old") -> AgentConfigDra
 
 def _client(*, snapshot_ref: str = "snapshot-ref-1") -> MagicMock:
     client = MagicMock()
+    client.initialize_home_snapshot_sync.return_value = SimpleNamespace(snapshot_ref=snapshot_ref)
     client.create_home_snapshot_from_binding_sync.return_value = SimpleNamespace(snapshot_ref=snapshot_ref)
     return client
 
 
-def test_validate_home_snapshot_binding_accepts_default_home_without_ledger_lookup() -> None:
+def test_create_initial_persists_backend_snapshot_ref(monkeypatch: pytest.MonkeyPatch) -> None:
     session = MagicMock()
-    validate_home_snapshot_binding(
+    client = _client()
+    monkeypatch.setattr(AgentHomeSnapshotService, "_client", lambda: nullcontext(client))
+
+    snapshot = AgentHomeSnapshotService.create_initial(
         session=session,
-        agent=Agent(id="agent-1"),
-        home_snapshot_id=None,
+        tenant_id="tenant-1",
+        agent_id="agent-1",
     )
 
-    session.scalar.assert_not_called()
+    assert snapshot.snapshot_ref == "snapshot-ref-1"
+    assert snapshot.status is AgentWorkingResourceStatus.ACTIVE
+    session.add.assert_called_once_with(snapshot)
+    session.flush.assert_called_once_with()
+
+
+def test_create_initial_flush_failure_does_not_delete_backend_snapshot(monkeypatch: pytest.MonkeyPatch) -> None:
+    session = MagicMock()
+    session.flush.side_effect = RuntimeError("flush failed")
+    client = _client()
+    monkeypatch.setattr(AgentHomeSnapshotService, "_client", lambda: nullcontext(client))
+
+    with pytest.raises(RuntimeError, match="flush failed"):
+        AgentHomeSnapshotService.create_initial(
+            session=session,
+            tenant_id="tenant-1",
+            agent_id="agent-1",
+        )
+
+    client.delete_home_snapshot_sync.assert_not_called()
 
 
 @pytest.mark.parametrize(
@@ -76,8 +98,7 @@ def test_build_apply_checkpoints_exact_active_binding(
     client = _client(snapshot_ref="snapshot-ref-2")
     monkeypatch.setattr(AgentHomeSnapshotService, "_client", lambda: nullcontext(client))
     monkeypatch.setattr(AgentWorkspaceService, "get_active_binding", get_binding)
-    validate_generation = MagicMock()
-    monkeypatch.setattr(AgentWorkspaceService, "validate_binding_generation", validate_generation)
+    monkeypatch.setattr(AgentWorkspaceService, "validate_binding_generation", MagicMock())
 
     snapshot = AgentHomeSnapshotService.create_for_build_apply(
         session=session,
@@ -89,32 +110,6 @@ def test_build_apply_checkpoints_exact_active_binding(
     request = client.create_home_snapshot_from_binding_sync.call_args.args[0]
     assert request.backend_binding_ref == "binding-ref-1"
     assert snapshot.snapshot_ref == "snapshot-ref-2"
-    assert validate_generation.call_args.kwargs["base_home_snapshot_id"] == "home-old"
-
-
-def test_build_apply_forwards_default_home_generation(monkeypatch: pytest.MonkeyPatch) -> None:
-    session = MagicMock()
-    session.scalar.return_value = SimpleNamespace(app_id="app-1", backing_app_id=None)
-    binding = SimpleNamespace(
-        backend_binding_ref="binding-ref-1",
-        agent_id="agent-1",
-        base_home_snapshot_id=None,
-        agent_config_version_id="build-1",
-        agent_config_version_kind="build_draft",
-    )
-    client = _client(snapshot_ref="snapshot-ref-2")
-    validate_generation = MagicMock()
-    monkeypatch.setattr(AgentHomeSnapshotService, "_client", lambda: nullcontext(client))
-    monkeypatch.setattr(AgentWorkspaceService, "get_active_binding", MagicMock(return_value=binding))
-    monkeypatch.setattr(AgentWorkspaceService, "validate_binding_generation", validate_generation)
-
-    snapshot = AgentHomeSnapshotService.create_for_build_apply(
-        session=session,
-        build_draft=_build_draft(home_snapshot_id=None),
-    )
-
-    assert snapshot.snapshot_ref == "snapshot-ref-2"
-    assert validate_generation.call_args.kwargs["base_home_snapshot_id"] is None
 
 
 def test_build_apply_fails_fast_without_source_binding() -> None:

api/tests/unit_tests/services/agent/test_workspace_service.py

@@ -15,7 +15,7 @@ from models.agent import (
     AgentWorkspaceBinding,
     AgentWorkspaceOwnerType,
 )
-from services.agent.workspace_service import AgentWorkspaceNotFoundError, AgentWorkspaceService, WorkspaceOwnerScope
+from services.agent.workspace_service import AgentWorkspaceService, WorkspaceOwnerScope
 
 
 def _scope() -> WorkspaceOwnerScope:
@@ -127,59 +127,6 @@ def test_create_binding_success_persists_new_workspace_and_binding(
     request = client.create_execution_binding_sync.call_args.args[0]
     assert request.existing_workspace_ref is None
     assert request.workspace_id == workspace.id
-    assert request.home_snapshot_ref == "home-ref"
-
-
-@pytest.mark.parametrize(
-    "sqlite_session",
-    [(AgentHomeSnapshot, AgentWorkspace, AgentWorkspaceBinding)],
-    indirect=True,
-)
-def test_create_binding_without_home_snapshot_uses_backend_default(
-    monkeypatch: pytest.MonkeyPatch, sqlite_session: Session
-) -> None:
-    client = _backend_client()
-    monkeypatch.setattr(AgentWorkspaceService, "_client", lambda: nullcontext(client))
-
-    binding = AgentWorkspaceService.create_binding(
-        session=sqlite_session,
-        scope=_scope(),
-        agent_id="agent-1",
-        base_home_snapshot_id=None,
-        agent_config_version_id="config-1",
-        agent_config_version_kind=AgentConfigVersionKind.SNAPSHOT,
-    )
-    sqlite_session.commit()
-
-    stored_binding = sqlite_session.get(AgentWorkspaceBinding, binding.id)
-    assert stored_binding is not None
-    assert stored_binding.base_home_snapshot_id is None
-    request = client.create_execution_binding_sync.call_args.args[0]
-    assert request.home_snapshot_ref is None
-
-
-@pytest.mark.parametrize(
-    "sqlite_session",
-    [(AgentHomeSnapshot, AgentWorkspace, AgentWorkspaceBinding)],
-    indirect=True,
-)
-def test_create_binding_rejects_missing_explicit_home_snapshot_before_backend_call(
-    monkeypatch: pytest.MonkeyPatch, sqlite_session: Session
-) -> None:
-    client = _backend_client()
-    monkeypatch.setattr(AgentWorkspaceService, "_client", lambda: nullcontext(client))
-
-    with pytest.raises(AgentWorkspaceNotFoundError, match="base Home Snapshot is unavailable"):
-        AgentWorkspaceService.create_binding(
-            session=sqlite_session,
-            scope=_scope(),
-            agent_id="agent-1",
-            base_home_snapshot_id="missing-home",
-            agent_config_version_id="config-1",
-            agent_config_version_kind=AgentConfigVersionKind.SNAPSHOT,
-        )
-
-    client.create_execution_binding_sync.assert_not_called()
 
 
 @pytest.mark.parametrize(

api/tests/unit_tests/services/test_human_input_service.py

@@ -4,6 +4,7 @@ from datetime import datetime, timedelta
 from unittest.mock import MagicMock
 
 import pytest
+from pydantic import JsonValue
 from pytest_mock import MockerFixture
 from sqlalchemy.engine import Engine
 from sqlalchemy.orm import Session, sessionmaker
@@ -40,12 +41,6 @@ from services.human_input_service import (
 )
 
 
-@pytest.fixture
-def unbound_session_factory() -> sessionmaker[Session]:
-    """Supply the required constructor dependency without enabling database access."""
-    return sessionmaker()
-
-
 def _make_app(mode: AppMode) -> App:
     return App(
         id="app-id",
@@ -61,7 +56,7 @@ def _make_app(mode: AppMode) -> App:
 
 
 @pytest.fixture
-def sample_form_record():
+def sample_form_record() -> HumanInputFormRecord:
     return HumanInputFormRecord(
         form_id="form-id",
         workflow_run_id="workflow-run-id",
@@ -95,7 +90,7 @@ def sample_form_record():
 def test_enqueue_resume_dispatches_task_for_workflow(
     mocker: MockerFixture,
     sqlite_session_factory: sessionmaker[Session],
-):
+) -> None:
     service = HumanInputService(sqlite_session_factory)
 
     workflow_run = MagicMock()
@@ -121,8 +116,10 @@ def test_enqueue_resume_dispatches_task_for_workflow(
 
 
 def test_ensure_form_active_respects_global_timeout(
-    monkeypatch, sample_form_record: HumanInputFormRecord, unbound_session_factory
-):
+    monkeypatch: pytest.MonkeyPatch,
+    sample_form_record: HumanInputFormRecord,
+    unbound_session_factory: sessionmaker[Session],
+) -> None:
     service = HumanInputService(unbound_session_factory)
     expired_record = dataclasses.replace(
         sample_form_record,
@@ -138,7 +135,7 @@ def test_ensure_form_active_respects_global_timeout(
 def test_enqueue_resume_dispatches_task_for_advanced_chat(
     mocker: MockerFixture,
     sqlite_session_factory: sessionmaker[Session],
-):
+) -> None:
     service = HumanInputService(sqlite_session_factory)
 
     workflow_run = MagicMock()
@@ -166,7 +163,7 @@ def test_enqueue_resume_dispatches_task_for_advanced_chat(
 def test_enqueue_resume_skips_unsupported_app_mode(
     mocker: MockerFixture,
     sqlite_session_factory: sessionmaker[Session],
-):
+) -> None:
     service = HumanInputService(sqlite_session_factory)
 
     workflow_run = MagicMock()
@@ -190,8 +187,9 @@ def test_enqueue_resume_skips_unsupported_app_mode(
 
 
 def test_get_form_definition_by_token_for_console_uses_repository(
-    sample_form_record: HumanInputFormRecord, unbound_session_factory
-):
+    sample_form_record: HumanInputFormRecord,
+    unbound_session_factory: sessionmaker[Session],
+) -> None:
     repo = MagicMock(spec=HumanInputFormSubmissionRepository)
     console_record = dataclasses.replace(sample_form_record, recipient_type=RecipientType.CONSOLE)
     repo.get_by_token.return_value = console_record
@@ -232,8 +230,10 @@ def _build_resumption_context_state(*, options: list[str], workflow_run_id: str)
 
 
 def test_resolve_form_inputs_uses_runtime_select_options(
-    sample_form_record: HumanInputFormRecord, unbound_session_factory, mocker: MockerFixture
-):
+    sample_form_record: HumanInputFormRecord,
+    unbound_session_factory: sessionmaker[Session],
+    mocker: MockerFixture,
+) -> None:
     configured_input = SelectInputConfig(
         output_variable_name="decision",
         option_source=StringListSource(
@@ -270,8 +270,10 @@ def test_resolve_form_inputs_uses_runtime_select_options(
 
 
 def test_submit_form_by_token_calls_repository_and_enqueue(
-    sample_form_record: HumanInputFormRecord, unbound_session_factory, mocker: MockerFixture
-):
+    sample_form_record: HumanInputFormRecord,
+    unbound_session_factory: sessionmaker[Session],
+    mocker: MockerFixture,
+) -> None:
     repo = MagicMock(spec=HumanInputFormSubmissionRepository)
     repo.get_by_token.return_value = sample_form_record
     repo.mark_submitted.return_value = sample_form_record
@@ -298,8 +300,10 @@ def test_submit_form_by_token_calls_repository_and_enqueue(
 
 
 def test_submit_form_by_token_enqueues_agent_app_resume_for_conversation_form(
-    sample_form_record, unbound_session_factory, mocker: MockerFixture
-):
+    sample_form_record: HumanInputFormRecord,
+    unbound_session_factory: sessionmaker[Session],
+    mocker: MockerFixture,
+) -> None:
     # ENG-635: a conversation-owned (Agent v2 chat) form routes to the chat
     # resume, not the workflow resume.
     repo = MagicMock(spec=HumanInputFormSubmissionRepository)
@@ -327,8 +331,10 @@ def test_submit_form_by_token_enqueues_agent_app_resume_for_conversation_form(
 
 
 def test_submit_form_by_token_skips_enqueue_for_delivery_test(
-    sample_form_record: HumanInputFormRecord, unbound_session_factory, mocker: MockerFixture
-):
+    sample_form_record: HumanInputFormRecord,
+    unbound_session_factory: sessionmaker[Session],
+    mocker: MockerFixture,
+) -> None:
     repo = MagicMock(spec=HumanInputFormSubmissionRepository)
     test_record = dataclasses.replace(
         sample_form_record,
@@ -351,8 +357,10 @@ def test_submit_form_by_token_skips_enqueue_for_delivery_test(
 
 
 def test_submit_form_by_token_passes_submission_user_id(
-    sample_form_record: HumanInputFormRecord, unbound_session_factory, mocker: MockerFixture
-):
+    sample_form_record: HumanInputFormRecord,
+    unbound_session_factory: sessionmaker[Session],
+    mocker: MockerFixture,
+) -> None:
     repo = MagicMock(spec=HumanInputFormSubmissionRepository)
     repo.get_by_token.return_value = sample_form_record
     repo.mark_submitted.return_value = sample_form_record
@@ -373,7 +381,10 @@ def test_submit_form_by_token_passes_submission_user_id(
     enqueue_spy.assert_called_once_with(sample_form_record.workflow_run_id)
 
 
-def test_submit_form_by_token_invalid_action(sample_form_record: HumanInputFormRecord, unbound_session_factory):
+def test_submit_form_by_token_invalid_action(
+    sample_form_record: HumanInputFormRecord,
+    unbound_session_factory: sessionmaker[Session],
+) -> None:
     repo = MagicMock(spec=HumanInputFormSubmissionRepository)
     repo.get_by_token.return_value = dataclasses.replace(sample_form_record)
     service = HumanInputService(unbound_session_factory, form_repository=repo)
@@ -390,7 +401,10 @@ def test_submit_form_by_token_invalid_action(sample_form_record: HumanInputFormR
     repo.mark_submitted.assert_not_called()
 
 
-def test_submit_form_by_token_missing_inputs(sample_form_record: HumanInputFormRecord, unbound_session_factory):
+def test_submit_form_by_token_missing_inputs(
+    sample_form_record: HumanInputFormRecord,
+    unbound_session_factory: sessionmaker[Session],
+) -> None:
     repo = MagicMock(spec=HumanInputFormSubmissionRepository)
 
     definition_with_input = FormDefinition(
@@ -465,12 +479,12 @@ def test_submit_form_by_token_missing_inputs(sample_form_record: HumanInputFormR
     ],
 )
 def test_validate_human_input_submission_rejects_invalid_select_and_file_payloads(
-    sample_form_record,
-    unbound_session_factory,
-    input_definition,
-    submitted_value,
-    expected_message,
-):
+    sample_form_record: HumanInputFormRecord,
+    unbound_session_factory: sessionmaker[Session],
+    input_definition: dict[str, JsonValue],
+    submitted_value: JsonValue,
+    expected_message: str,
+) -> None:
     repo = MagicMock(spec=HumanInputFormSubmissionRepository)
     definition = FormDefinition.model_validate(
         {
@@ -489,14 +503,14 @@ def test_validate_human_input_submission_rejects_invalid_select_and_file_payload
             recipient_type=RecipientType.STANDALONE_WEB_APP,
             form_token="token",
             selected_action_id="submit",
-            form_data={input_definition["output_variable_name"]: submitted_value},
+            form_data={definition.inputs[0].output_variable_name: submitted_value},
         )
 
     assert expected_message in str(exc_info.value)
     repo.mark_submitted.assert_not_called()
 
 
-def test_form_properties(sample_form_record: HumanInputFormRecord):
+def test_form_properties(sample_form_record: HumanInputFormRecord) -> None:
     form = Form(sample_form_record)
     assert form.id == "form-id"
     assert form.workflow_run_id == "workflow-run-id"
@@ -510,20 +524,20 @@ def test_form_properties(sample_form_record: HumanInputFormRecord):
     assert isinstance(form.expiration_time, datetime)
 
 
-def test_form_submitted_error_init():
+def test_form_submitted_error_init() -> None:
     error = FormSubmittedError(form_id="test-form")
     assert error.description == "This form has already been submitted by another user, form_id=test-form"
     assert error.code == 412
 
 
-def test_human_input_service_init_with_engine(sqlite_engine: Engine):
+def test_human_input_service_init_with_engine(sqlite_engine: Engine) -> None:
     service = HumanInputService(session_factory=sqlite_engine)
 
     assert isinstance(service._session_factory, sessionmaker)
     assert service._session_factory.kw["bind"] is sqlite_engine
 
 
-def test_get_form_by_token_none(unbound_session_factory):
+def test_get_form_by_token_none(unbound_session_factory: sessionmaker[Session]) -> None:
     repo = MagicMock(spec=HumanInputFormSubmissionRepository)
     repo.get_by_token.return_value = None
 
@@ -531,7 +545,10 @@ def test_get_form_by_token_none(unbound_session_factory):
     assert service.get_form_by_token("invalid") is None
 
 
-def test_get_form_definition_by_token_mismatch(sample_form_record: HumanInputFormRecord, unbound_session_factory):
+def test_get_form_definition_by_token_mismatch(
+    sample_form_record: HumanInputFormRecord,
+    unbound_session_factory: sessionmaker[Session],
+) -> None:
     repo = MagicMock(spec=HumanInputFormSubmissionRepository)
     repo.get_by_token.return_value = sample_form_record
 
@@ -540,7 +557,10 @@ def test_get_form_definition_by_token_mismatch(sample_form_record: HumanInputFor
     assert service.get_form_definition_by_token(RecipientType.CONSOLE, "token") is None
 
 
-def test_get_form_definition_by_token_success(sample_form_record: HumanInputFormRecord, unbound_session_factory):
+def test_get_form_definition_by_token_success(
+    sample_form_record: HumanInputFormRecord,
+    unbound_session_factory: sessionmaker[Session],
+) -> None:
     repo = MagicMock(spec=HumanInputFormSubmissionRepository)
     repo.get_by_token.return_value = sample_form_record
 
@@ -551,8 +571,9 @@ def test_get_form_definition_by_token_success(sample_form_record: HumanInputForm
 
 
 def test_get_form_definition_by_token_for_console_mismatch(
-    sample_form_record: HumanInputFormRecord, unbound_session_factory
-):
+    sample_form_record: HumanInputFormRecord,
+    unbound_session_factory: sessionmaker[Session],
+) -> None:
     repo = MagicMock(spec=HumanInputFormSubmissionRepository)
     repo.get_by_token.return_value = sample_form_record  # is STANDALONE_WEB_APP
 
@@ -560,7 +581,9 @@ def test_get_form_definition_by_token_for_console_mismatch(
     assert service.get_form_definition_by_token_for_console("token") is None
 
 
-def test_submit_form_by_token_delivery_not_enabled(unbound_session_factory):
+def test_submit_form_by_token_delivery_not_enabled(
+    unbound_session_factory: sessionmaker[Session],
+) -> None:
     repo = MagicMock(spec=HumanInputFormSubmissionRepository)
     repo.get_by_token.return_value = None
 
@@ -570,8 +593,10 @@ def test_submit_form_by_token_delivery_not_enabled(unbound_session_factory):
 
 
 def test_submit_form_by_token_no_workflow_run_id(
-    sample_form_record: HumanInputFormRecord, unbound_session_factory, mocker: MockerFixture
-):
+    sample_form_record: HumanInputFormRecord,
+    unbound_session_factory: sessionmaker[Session],
+    mocker: MockerFixture,
+) -> None:
     repo = MagicMock(spec=HumanInputFormSubmissionRepository)
     repo.get_by_token.return_value = sample_form_record
 
@@ -586,7 +611,10 @@ def test_submit_form_by_token_no_workflow_run_id(
     enqueue_spy.assert_not_called()
 
 
-def test_ensure_form_active_errors(sample_form_record: HumanInputFormRecord, unbound_session_factory):
+def test_ensure_form_active_errors(
+    sample_form_record: HumanInputFormRecord,
+    unbound_session_factory: sessionmaker[Session],
+) -> None:
     service = HumanInputService(unbound_session_factory)
 
     # Submitted
@@ -607,7 +635,10 @@ def test_ensure_form_active_errors(sample_form_record: HumanInputFormRecord, unb
         service.ensure_form_active(Form(expired_time_record))
 
 
-def test_ensure_not_submitted_raises(sample_form_record: HumanInputFormRecord, unbound_session_factory):
+def test_ensure_not_submitted_raises(
+    sample_form_record: HumanInputFormRecord,
+    unbound_session_factory: sessionmaker[Session],
+) -> None:
     service = HumanInputService(unbound_session_factory)
     submitted_record = dataclasses.replace(sample_form_record, submitted_at=naive_utc_now())
 
@@ -615,7 +646,10 @@ def test_ensure_not_submitted_raises(sample_form_record: HumanInputFormRecord, u
         service._ensure_not_submitted(Form(submitted_record))
 
 
-def test_enqueue_resume_workflow_not_found(mocker: MockerFixture, unbound_session_factory):
+def test_enqueue_resume_workflow_not_found(
+    mocker: MockerFixture,
+    unbound_session_factory: sessionmaker[Session],
+) -> None:
     service = HumanInputService(unbound_session_factory)
 
     workflow_run_repo = MagicMock()
@@ -631,10 +665,10 @@ def test_enqueue_resume_workflow_not_found(mocker: MockerFixture, unbound_sessio
 
 
 def test_enqueue_resume_app_not_found(
-    mocker,
+    mocker: MockerFixture,
     sqlite_session_factory: sessionmaker[Session],
     caplog: pytest.LogCaptureFixture,
-):
+) -> None:
     service = HumanInputService(sqlite_session_factory)
 
     workflow_run = MagicMock()
@@ -660,8 +694,10 @@ def test_enqueue_resume_app_not_found(
 
 
 def test_is_globally_expired_zero_timeout(
-    monkeypatch: pytest.MonkeyPatch, sample_form_record: HumanInputFormRecord, unbound_session_factory
-):
+    monkeypatch: pytest.MonkeyPatch,
+    sample_form_record: HumanInputFormRecord,
+    unbound_session_factory: sessionmaker[Session],
+) -> None:
     service = HumanInputService(unbound_session_factory)
 
     monkeypatch.setattr(human_input_service_module.dify_config, "HUMAN_INPUT_GLOBAL_TIMEOUT_SECONDS", 0)
@@ -669,7 +705,9 @@ def test_is_globally_expired_zero_timeout(
 
 
 def test_submit_form_by_token_normalizes_select_and_files(
-    sample_form_record: HumanInputFormRecord, unbound_session_factory, mocker: MockerFixture
+    sample_form_record: HumanInputFormRecord,
+    unbound_session_factory: sessionmaker[Session],
+    mocker: MockerFixture,
 ) -> None:
     repo = MagicMock(spec=HumanInputFormSubmissionRepository)
     definition = FormDefinition(
@@ -751,7 +789,8 @@ def test_submit_form_by_token_normalizes_select_and_files(
 
 
 def test_submit_form_by_token_invalid_select_value(
-    sample_form_record: HumanInputFormRecord, unbound_session_factory
+    sample_form_record: HumanInputFormRecord,
+    unbound_session_factory: sessionmaker[Session],
 ) -> None:
     repo = MagicMock(spec=HumanInputFormSubmissionRepository)
     definition = FormDefinition(
@@ -779,7 +818,8 @@ def test_submit_form_by_token_invalid_select_value(
 
 
 def test_submit_form_by_token_invalid_file_list_item(
-    sample_form_record: HumanInputFormRecord, unbound_session_factory
+    sample_form_record: HumanInputFormRecord,
+    unbound_session_factory: sessionmaker[Session],
 ) -> None:
     repo = MagicMock(spec=HumanInputFormSubmissionRepository)
     definition = FormDefinition(
@@ -805,7 +845,9 @@ def test_submit_form_by_token_invalid_file_list_item(
 
 
 def test_submit_form_by_token_rejects_cross_tenant_file(
-    sample_form_record: HumanInputFormRecord, unbound_session_factory, mocker: MockerFixture
+    sample_form_record: HumanInputFormRecord,
+    unbound_session_factory: sessionmaker[Session],
+    mocker: MockerFixture,
 ) -> None:
     repo = MagicMock(spec=HumanInputFormSubmissionRepository)
     definition = FormDefinition(
@@ -837,7 +879,9 @@ def test_submit_form_by_token_rejects_cross_tenant_file(
 
 
 def test_submit_form_by_token_rejects_cross_tenant_file_list(
-    sample_form_record: HumanInputFormRecord, unbound_session_factory, mocker: MockerFixture
+    sample_form_record: HumanInputFormRecord,
+    unbound_session_factory: sessionmaker[Session],
+    mocker: MockerFixture,
 ) -> None:
     repo = MagicMock(spec=HumanInputFormSubmissionRepository)
     definition = FormDefinition(

api/tests/unit_tests/test_sqlite_fixtures.py

@@ -5,6 +5,7 @@ from threading import Barrier
 import pytest
 from sqlalchemy import create_engine, inspect, text
 from sqlalchemy.engine import URL, Engine
+from sqlalchemy.exc import UnboundExecutionError
 from sqlalchemy.orm import Session, sessionmaker
 from sqlalchemy.pool import QueuePool
 
@@ -58,6 +59,25 @@ def test_core_session_factory_uses_the_shared_sqlite_session_factory(
         assert session.scalar(text("SELECT value FROM global_factory_probe")) == 42
 
 
+def test_unbound_session_factory_disables_explicit_and_global_database_access(
+    unbound_session_factory: sessionmaker[Session],
+) -> None:
+    assert session_factory_module.session_factory.get_session_maker() is unbound_session_factory
+
+    with unbound_session_factory() as session:
+        with pytest.raises(UnboundExecutionError):
+            session.get_bind()
+
+    with session_factory_module.session_factory.create_session() as session:
+        with pytest.raises(UnboundExecutionError):
+            session.execute(text("SELECT 1"))
+
+
+def test_unbound_session_rejects_database_access(unbound_session: Session) -> None:
+    with pytest.raises(UnboundExecutionError):
+        unbound_session.scalar(text("SELECT 1"))
+
+
 def test_sqlite_session_factory_shares_one_database_across_worker_sessions(
     sqlite_session_factory: sessionmaker[Session],
 ) -> None:

dify-agent/docs/dify-agent/concepts/runtime-resources/index.md

@@ -67,17 +67,12 @@ streams are observability state, not the Home/Workspace/Binding ledger.
 
 ## Creation and execution flow
 
-Agent creation does not create a Home Snapshot. A config with no logical Home
-Snapshot asks the selected backend to materialize its deployment-default Home
-when the Binding is created. This default Home is mutable and private to the
-Binding; it does not produce an `agent_home_snapshots` row or an implicit
-snapshot ref.
-
-Build Draft Apply uses `POST /home-snapshots/from-binding`: Dify Agent acquires
-the exact source Binding, snapshots its materialized Home through the
-backend-native operation, releases the lease, and returns a new opaque snapshot
-ref. Dify API then stores a new immutable `agent_home_snapshots` row and records
-its logical id on the resulting config version. There is no replay or fallback
+Home Snapshot initialization uses `POST /home-snapshots/initialize`. Build Draft
+Apply uses `POST /home-snapshots/from-binding`: Dify Agent acquires the exact
+source Binding, snapshots its materialized Home through the backend-native
+operation, releases the lease, and returns a new opaque snapshot ref. Dify API
+then stores a new immutable `agent_home_snapshots` row and records its logical id
+on the resulting config version. There is no replay or initialization fallback
 when the source Binding is unavailable.
 
 Before an Agent request, Dify API loads the specific product context. If it has
@@ -87,12 +82,10 @@ its owner and config/Home generation. Missing, retired, or mismatched Bindings
 fail fast; Dify API does not search by Agent, Workspace, candidate count, or
 recency, and it does not create a replacement implicitly.
 
-`POST /execution-bindings` accepts either an exact `home_snapshot_ref` or
-`null`. An exact ref must be materialized without fallback; `null` selects the
-backend's deployment-default Home. It returns opaque Binding and Workspace
-refs. Every create request represents a new participant, even when the Agent,
-Snapshot, config generation, and Workspace match another Binding. The request
-composition contains:
+`POST /execution-bindings` materializes the selected Home Snapshot and returns
+opaque Binding and Workspace refs. Every create request represents a new
+participant, even when the Agent, Snapshot, config generation, and Workspace
+match another Binding. The request composition contains:
 
 ```json
 {
@@ -181,9 +174,9 @@ own Home plus the shared Workspace.
 
 | Backend | Home Snapshot operations | Binding operations | Physical relationship |
 | --- | --- | --- | --- |
-| Local | Supported | Supported, including default empty Homes and attaching multiple Bindings to one Workspace | Snapshot directory, per-Binding materialized Home, and Workspace directory are separate. |
-| E2B | Supported | Supported with template-backed default Homes, without shared-Workspace attachment | Binding and Workspace refs map to the same E2B resource; checkpoints use E2B snapshots. |
-| Enterprise | Not implemented | Default-Home Binding creation, acquire, and coupled destroy are supported | Binding and Workspace refs map to one Gateway sandbox. Explicit Home Snapshot materialization fails fast. |
+| Local | Supported | Supported, including attaching multiple Bindings to one Workspace | Snapshot directory, per-Binding materialized Home, and Workspace directory are separate. |
+| E2B | Supported | Supported without shared-Workspace attachment | Binding and Workspace refs map to the same E2B resource; Home initialization/checkpoint uses E2B snapshots. |
+| Enterprise | Not implemented | Not implemented | Configuration is accepted, but every resource operation fails fast with `NotImplementedError`. |
 
 Local creates a new Home for every Binding id. Destroying one Binding without
 the Workspace leaves sibling Homes and the shared Workspace intact. Current E2B
@@ -192,8 +185,9 @@ its Binding and Workspace are one Sandbox. It also rejects binding-only destroy.
 Neither path creates a fallback Workspace or switches backends.
 
 `DIFY_AGENT_E2B_ACTIVE_TIMEOUT_SECONDS` limits continuous active time for an E2B
-resource. Runtime resources pause on timeout. It is not a retention TTL and
-does not delete paused resources or immutable snapshots.
+resource. Runtime resources pause on timeout; temporary Home initialization
+resources are killed. It is not a retention TTL and does not delete paused
+resources or immutable snapshots.
 
 See the [Shell layer](../../user-manual/shell-layer/index.md) for request
 composition and the [Operations Guide](../../guide/index.md) for Local and E2B

dify-agent/docs/dify-agent/get-started/index.md

@@ -85,9 +85,10 @@ DIFY_AGENT_SANDBOX_FILE_UPLOAD_MAX_BYTES=52428800
 
 E2B requires `DIFY_AGENT_E2B_API_KEY` and defaults to the prepared
 `difys-default-team/dify-agent-local-sandbox` template. The E2B active timeout
-pauses the physical resource behind a Binding; it is not a retention TTL.
-Enterprise supports Bindings created from its deployment-default Home.
-Immutable Home Snapshot creation and materialization remain unsupported there.
+pauses the physical resource behind a Binding or kills temporary Home
+initialization resources; it is not a retention TTL. Enterprise settings are
+accepted, but current Home Snapshot and Binding operations fail fast with
+`NotImplementedError`.
 
 A shell-enabled request includes Execution Context, `dify.runtime`, and
 `dify.shell`. Dify API creates or resolves the specific persistent Binding for

dify-agent/docs/dify-agent/guide/index.md

@@ -42,13 +42,13 @@ also reads `.env` and `dify-agent/.env` when present.
 | `DIFY_AGENT_RUNTIME_BACKEND` | `local` | Selects one coherent `local`, `enterprise`, or `e2b` Home Snapshot + Execution Binding backend profile. |
 | `DIFY_AGENT_LOCAL_SANDBOX_ENDPOINT` | empty | Local shellctl data-plane URL. With the default Local selection, leaving it empty disables `dify.runtime` and resource endpoints. |
 | `DIFY_AGENT_LOCAL_SANDBOX_AUTH_TOKEN` | empty | Optional bearer token sent to Local shellctl. |
-| `DIFY_AGENT_ENTERPRISE_SANDBOX_GATEWAY_ENDPOINT` | empty | Enterprise Gateway endpoint required by configuration. Default-Home Bindings are supported; immutable Home Snapshot operations remain unsupported. |
+| `DIFY_AGENT_ENTERPRISE_SANDBOX_GATEWAY_ENDPOINT` | empty | Enterprise Gateway endpoint required by configuration. Current Home Snapshot and Binding operations fail fast with `NotImplementedError`. |
 | `DIFY_AGENT_ENTERPRISE_SANDBOX_GATEWAY_AUTH_TOKEN` | empty | Optional `X-Inner-Api-Key` sent to the Enterprise Gateway. |
 | `DIFY_AGENT_ENTERPRISE_SANDBOX_GATEWAY_TIMEOUT` | `30` | Enterprise control-plane timeout in seconds. |
 | `DIFY_AGENT_ENTERPRISE_SANDBOX_PROXY_TIMEOUT` | `60` | Enterprise shellctl-proxy timeout in seconds. |
 | `DIFY_AGENT_E2B_API_KEY` | empty | E2B API key; required for E2B. |
-| `DIFY_AGENT_E2B_TEMPLATE` | `difys-default-team/dify-agent-local-sandbox` | Prepared E2B template containing shellctl and the deployment-default Home environment. |
-| `DIFY_AGENT_E2B_ACTIVE_TIMEOUT_SECONDS` | `3600` | Maximum continuous active time, up to 3600 seconds. Binding resources pause on timeout. This is not a retention TTL. |
+| `DIFY_AGENT_E2B_TEMPLATE` | `difys-default-team/dify-agent-local-sandbox` | Prepared E2B template containing shellctl and the initial Home environment. |
+| `DIFY_AGENT_E2B_ACTIVE_TIMEOUT_SECONDS` | `3600` | Maximum continuous active time, up to 3600 seconds. Binding resources pause on timeout; temporary Home initialization resources are killed. This is not a retention TTL. |
 | `DIFY_AGENT_E2B_SHELLCTL_AUTH_TOKEN` | empty | Optional bearer token expected by shellctl inside the E2B template. |
 | `DIFY_AGENT_E2B_SHELLCTL_PORT` | `5004` | shellctl port exposed by the E2B template. |
 | `DIFY_AGENT_SANDBOX_FILE_UPLOAD_MAX_BYTES` | `52428800` | Standalone Dify Agent maximum for whole-file Workspace upload capture; 50 MiB by default. Docker Compose derives it from `PLUGIN_MAX_FILE_SIZE`. |
@@ -177,9 +177,12 @@ docker compose \
 ```
 
 `DIFY_AGENT_E2B_ACTIVE_TIMEOUT_SECONDS` controls continuous active E2B time.
-The physical resource behind a Binding pauses when that timeout fires, preserving
-the current Workspace. The setting is not a resource-age TTL and does not delete
-paused resources or immutable snapshots.
+The physical resource behind a Binding pauses when that timeout fires; a
+temporary Home initialization resource is killed. Pausing preserves the current
+Workspace. The setting does not delete an aged paused resource or immutable
+snapshot, and Dify Agent currently has no resource-age TTL, reconciler, or
+eventual cleanup guarantee. Dify API retirement followed by Binding collection
+kills the coupled E2B resource.
 
 ## Run runtime-backend integration contracts
 

dify-agent/docs/dify-agent/user-manual/shell-layer/index.md

@@ -41,10 +41,10 @@ the opaque Binding ref belongs to `DifyRuntimeLayerConfig`.
 ## Runtime requirements
 
 The server constructs one coherent runtime backend profile. Local and E2B
-implement Home Snapshot and Execution Binding operations. Enterprise implements
-default-Home Binding creation, acquisition, and coupled destruction, while
-immutable Home Snapshot operations fail fast; there is no compatibility
-fallback to the retired Sandbox protocol.
+implement Home Snapshot and Execution Binding operations. Enterprise settings
+can be selected, but resource operations currently fail fast with
+`NotImplementedError`; there is no compatibility fallback to the retired
+Sandbox protocol.
 
 ```python
 from dify_agent.runtime.compositor_factory import create_default_layer_providers

dify-agent/src/dify_agent/client/_client.py

@@ -36,6 +36,7 @@ from dify_agent.protocol import (
     DeleteHomeSnapshotRequest,
     DestroyExecutionBindingRequest,
     HomeSnapshotResponse,
+    InitializeHomeSnapshotRequest,
     RUN_EVENT_ADAPTER,
     RunEvent,
     RunEventsResponse,
@@ -528,6 +529,24 @@ class Client:
         response = self._post_sync_json("destroy_execution_binding_sync", "/execution-bindings/destroy", request)
         _raise_for_status(response)
 
+    async def initialize_home_snapshot(self, request: InitializeHomeSnapshotRequest) -> HomeSnapshotResponse:
+        """Create a backend-native initial Home Snapshot."""
+        response = await self._post_async_json(
+            "initialize_home_snapshot",
+            "/home-snapshots/initialize",
+            request,
+        )
+        return _parse_model_response(response, HomeSnapshotResponse)
+
+    def initialize_home_snapshot_sync(self, request: InitializeHomeSnapshotRequest) -> HomeSnapshotResponse:
+        """Synchronous variant of ``initialize_home_snapshot``."""
+        response = self._post_sync_json(
+            "initialize_home_snapshot_sync",
+            "/home-snapshots/initialize",
+            request,
+        )
+        return _parse_model_response(response, HomeSnapshotResponse)
+
     async def create_home_snapshot_from_binding(
         self,
         request: CreateHomeSnapshotFromBindingRequest,

dify-agent/src/dify_agent/protocol/__init__.py

@@ -46,6 +46,7 @@ from .home_snapshot import (
     CreateHomeSnapshotFromBindingRequest,
     DeleteHomeSnapshotRequest,
     HomeSnapshotResponse,
+    InitializeHomeSnapshotRequest,
 )
 from .workspace import (
     WorkspaceFileEntry,
@@ -78,6 +79,7 @@ __all__ = [
     "EmptyRunEventData",
     "LayerExitSignals",
     "HomeSnapshotResponse",
+    "InitializeHomeSnapshotRequest",
     "PydanticAIStreamRunEvent",
     "RUN_EVENT_ADAPTER",
     "RunCancelledEvent",

dify-agent/src/dify_agent/protocol/execution_binding.py

@@ -11,7 +11,7 @@ class CreateExecutionBindingRequest(BaseModel):
     binding_id: str = Field(min_length=1)
     workspace_id: str = Field(min_length=1)
     existing_workspace_ref: str | None = None
-    home_snapshot_ref: str | None = Field(default=None, min_length=1)
+    home_snapshot_ref: str = Field(min_length=1)
 
     model_config: ClassVar[ConfigDict] = ConfigDict(extra="forbid")
 

dify-agent/src/dify_agent/protocol/home_snapshot.py

@@ -5,6 +5,14 @@ from typing import ClassVar
 from pydantic import BaseModel, ConfigDict, Field
 
 
+class InitializeHomeSnapshotRequest(BaseModel):
+    tenant_id: str = Field(min_length=1)
+    agent_id: str = Field(min_length=1)
+    home_snapshot_id: str = Field(min_length=1)
+
+    model_config: ClassVar[ConfigDict] = ConfigDict(extra="forbid")
+
+
 class CreateHomeSnapshotFromBindingRequest(BaseModel):
     tenant_id: str = Field(min_length=1)
     agent_id: str = Field(min_length=1)
@@ -30,4 +38,5 @@ __all__ = [
     "CreateHomeSnapshotFromBindingRequest",
     "DeleteHomeSnapshotRequest",
     "HomeSnapshotResponse",
+    "InitializeHomeSnapshotRequest",
 ]

dify-agent/src/dify_agent/runtime_backend/__init__.py

@@ -22,6 +22,7 @@ from .protocols import (
     FileSystem,
     HomeSnapshotBackend,
     HomeSnapshotCreateSpec,
+    InitializeHomeSnapshotSpec,
     RuntimeBackendProfile,
     RuntimeLayout,
     RuntimeLease,
@@ -45,6 +46,7 @@ __all__ = [
     "HomeSnapshotCreateError",
     "HomeSnapshotCreateSpec",
     "HomeSnapshotNotFoundError",
+    "InitializeHomeSnapshotSpec",
     "RuntimeBackendError",
     "RuntimeBackendProfile",
     "RuntimeLayout",

dify-agent/src/dify_agent/runtime_backend/e2b.py

@@ -30,6 +30,7 @@ from dify_agent.runtime_backend.protocols import (
     ExecutionBindingDestroySpec,
     FileSystem,
     HomeSnapshotCreateSpec,
+    InitializeHomeSnapshotSpec,
     RuntimeLayout,
     RuntimeLease,
 )
@@ -162,12 +163,44 @@ class E2BSDKControlPlane:
 class E2BHomeSnapshotBackend:
     """Implement immutable Home Snapshot operations with E2B snapshots.
 
-    Build Apply snapshots the E2B resource behind the supplied ``RuntimeLease``.
-    Dify API stores the returned value as an opaque backend ref; this adapter
-    keeps no cross-request state.
+    Initialization snapshots the prepared deployment template and releases its
+    temporary E2B resource. Build Apply snapshots the E2B resource behind the
+    supplied ``RuntimeLease``. Dify API stores the returned value as an opaque
+    backend ref; this adapter keeps no cross-request state.
     """
 
     control_plane: E2BControlPlane
+    template: str
+    active_timeout_seconds: int
+    home_dir: str = "/home/dify"
+
+    async def initialize(self, spec: InitializeHomeSnapshotSpec) -> str:
+        sandbox: _E2BSandbox | None = None
+        try:
+            sandbox = await self.control_plane.create(
+                self.template,
+                timeout=self.active_timeout_seconds,
+                metadata={
+                    "dify.resource": "home-snapshot-initialize",
+                    "dify.tenant_id": spec.tenant_id,
+                    "dify.agent_id": spec.agent_id,
+                    "dify.home_snapshot_id": spec.home_snapshot_id,
+                },
+                on_timeout="kill",
+            )
+            _ = await sandbox.files.make_dir(self.home_dir)
+            snapshot = await sandbox.create_snapshot()
+            return snapshot.snapshot_id
+        except BaseException as exc:
+            if isinstance(exc, Exception):
+                raise HomeSnapshotCreateError(str(exc)) from exc
+            raise
+        finally:
+            if sandbox is not None:
+                try:
+                    _ = await sandbox.kill()
+                except BaseException:
+                    pass
 
     async def create_from_runtime(self, *, spec: HomeSnapshotCreateSpec, source: RuntimeLease) -> str:
         """Create an immutable E2B snapshot from the source Binding's active lease."""
@@ -203,7 +236,6 @@ class E2BExecutionBindingBackend:
     """
 
     control_plane: E2BControlPlane
-    template: str
     active_timeout_seconds: int
     shellctl_auth_token: str = ""
     shellctl_port: int = 5004
@@ -212,13 +244,13 @@ class E2BExecutionBindingBackend:
     )
 
     async def create_binding(self, spec: ExecutionBindingCreateSpec) -> ExecutionBindingAllocation:
-        """Create one paused E2B resource from a snapshot or deployment template."""
+        """Create one paused E2B resource from an immutable Home Snapshot ref."""
         if spec.existing_workspace_ref is not None:
             raise SharedWorkspaceUnsupportedError("current E2B backend cannot attach to an existing Workspace")
         sandbox: _E2BSandbox | None = None
         try:
             sandbox = await self.control_plane.create(
-                self.template if spec.home_snapshot_ref is None else spec.home_snapshot_ref,
+                spec.home_snapshot_ref,
                 timeout=self.active_timeout_seconds,
                 metadata={
                     "dify.resource": "runtime-sandbox",
@@ -242,7 +274,7 @@ class E2BExecutionBindingBackend:
                 except BaseException:
                     pass
             if isinstance(exc, Exception):
-                if isinstance(exc, BindingCreateError):
+                if isinstance(exc, (BindingCreateError, SharedWorkspaceUnsupportedError)):
                     raise
                 raise BindingCreateError(str(exc)) from exc
             raise

dify-agent/src/dify_agent/runtime_backend/enterprise.py

@@ -1,10 +1,10 @@
 """Enterprise Gateway adapter for the working-environment protocol.
 
-The existing Gateway can allocate, reconnect to, and delete a sandbox, but it
-does not expose immutable Home Snapshot operations. One physical sandbox owns
-both the materialized Home and Workspace, so their cleanup is coupled. Runtime
-access remains operation-local and is routed through the Gateway's shellctl
-proxy.
+The existing Gateway can reconnect to and delete an already allocated sandbox,
+but it cannot materialize a Home Snapshot or create a protocol-compliant
+Binding. One physical sandbox owns both the materialized Home and Workspace, so
+their cleanup is coupled. Runtime access remains operation-local and is routed
+through the Gateway's shellctl proxy.
 """
 
 from __future__ import annotations
@@ -21,10 +21,8 @@ from dify_agent.adapters.shell.protocols import ShellCommandProtocol, ShellProvi
 from dify_agent.adapters.shell.shellctl import ShellctlClientProtocol, ShellctlCommands
 from dify_agent.runtime_backend.errors import (
     BindingAcquireError,
-    BindingCreateError,
     BindingDestroyError,
     BindingLostError,
-    SharedWorkspaceUnsupportedError,
     WorkspacePreservationUnsupportedError,
 )
 from dify_agent.runtime_backend.protocols import (
@@ -33,6 +31,7 @@ from dify_agent.runtime_backend.protocols import (
     ExecutionBindingDestroySpec,
     FileSystem,
     HomeSnapshotCreateSpec,
+    InitializeHomeSnapshotSpec,
     RuntimeLayout,
     RuntimeLease,
 )
@@ -46,13 +45,21 @@ logger = logging.getLogger(__name__)
 
 
 def _not_implemented() -> NotImplementedError:
-    return NotImplementedError("Enterprise Gateway does not implement immutable Home Snapshot operations")
+    return NotImplementedError("Enterprise Gateway does not implement the Execution Binding protocol")
 
 
 @dataclass(slots=True)
 class EnterpriseHomeSnapshotBackend:
     """Reject Home Snapshot operations until the Gateway exposes immutable snapshots."""
 
+    gateway_endpoint: str
+    auth_token: str
+    gateway_timeout: float = 30.0
+
+    async def initialize(self, spec: InitializeHomeSnapshotSpec) -> str:
+        del spec
+        raise _not_implemented()
+
     async def create_from_runtime(self, *, spec: HomeSnapshotCreateSpec, source: RuntimeLease) -> str:
         del spec, source
         raise _not_implemented()
@@ -64,7 +71,7 @@ class EnterpriseHomeSnapshotBackend:
 
 @dataclass(slots=True)
 class EnterpriseExecutionBindingBackend:
-    """Manage Gateway sandboxes as coupled physical Bindings and Workspaces."""
+    """Access and destroy legacy Gateway sandboxes as coupled physical Bindings."""
 
     gateway_endpoint: str
     auth_token: str
@@ -75,56 +82,8 @@ class EnterpriseExecutionBindingBackend:
     )
 
     async def create_binding(self, spec: ExecutionBindingCreateSpec) -> ExecutionBindingAllocation:
-        """Create a default Gateway sandbox and initialize its canonical layout."""
-        if spec.existing_workspace_ref is not None:
-            raise SharedWorkspaceUnsupportedError("current Enterprise backend cannot attach to an existing Workspace")
-        if spec.home_snapshot_ref is not None:
-            raise BindingCreateError("current Enterprise backend cannot materialize an immutable Home Snapshot")
-
-        sandbox_id: str | None = None
-        data_plane: ShellctlRuntimeLease | None = None
-        headers = {"X-Inner-Api-Key": self.auth_token} if self.auth_token else {}
-        try:
-            async with httpx.AsyncClient(
-                base_url=self.gateway_endpoint.rstrip("/"),
-                headers=headers,
-                timeout=httpx.Timeout(self.gateway_timeout),
-            ) as client:
-                response = await client.post("/v1/sandboxes", json={"tenantId": spec.tenant_id})
-                _ = response.raise_for_status()
-                payload = response.json()
-            sandbox_id_value = payload.get("sandboxId") if isinstance(payload, dict) else None
-            if not isinstance(sandbox_id_value, str) or not sandbox_id_value:
-                raise BindingCreateError("Enterprise Gateway returned an invalid sandbox id")
-            sandbox_id = sandbox_id_value
-
-            data_plane = await self._create_data_plane(sandbox_id)
-            result = await run_shellctl_control_command(
-                ShellctlCommands(client=data_plane.client),
-                "\n".join(
-                    [
-                        "set -eu",
-                        f"mkdir -p {shlex.quote(self.layout.home_dir)}",
-                        f"rm -rf -- {shlex.quote(self.layout.workspace_dir)}",
-                        f"mkdir -p {shlex.quote(self.layout.workspace_dir)}",
-                        f"chmod 700 {shlex.quote(self.layout.home_dir)} {shlex.quote(self.layout.workspace_dir)}",
-                    ]
-                ),
-            )
-            if result.exit_code != 0:
-                raise BindingCreateError(result.output)
-            await data_plane.close()
-            data_plane = None
-            return ExecutionBindingAllocation(binding_ref=sandbox_id, workspace_ref=sandbox_id)
-        except BaseException as exc:
-            await _close_best_effort(data_plane, binding_ref=sandbox_id or spec.binding_id)
-            if sandbox_id is not None:
-                await self._delete_sandbox_best_effort(sandbox_id)
-            if isinstance(exc, BindingCreateError):
-                raise
-            if isinstance(exc, Exception):
-                raise BindingCreateError(str(exc)) from exc
-            raise
+        del spec
+        raise _not_implemented()
 
     async def acquire(self, binding_ref: str) -> RuntimeLease:
         """Reconnect to one existing Gateway sandbox without creating a replacement."""
@@ -178,34 +137,21 @@ class EnterpriseExecutionBindingBackend:
         if spec.workspace_ref != spec.binding_ref:
             raise BindingDestroyError("Enterprise Workspace ref must equal its Binding ref")
 
+        headers = {"X-Inner-Api-Key": self.auth_token} if self.auth_token else {}
+        encoded_binding_ref = quote(spec.binding_ref, safe="")
         try:
-            await self._delete_sandbox(spec.binding_ref)
+            async with httpx.AsyncClient(
+                base_url=self.gateway_endpoint.rstrip("/"),
+                headers=headers,
+                timeout=httpx.Timeout(self.gateway_timeout),
+            ) as client:
+                response = await client.delete(f"/v1/sandboxes/{encoded_binding_ref}")
+                if response.status_code == 404:
+                    return
+                _ = response.raise_for_status()
         except (httpx.TimeoutException, httpx.RequestError, httpx.HTTPStatusError) as exc:
             raise BindingDestroyError(str(exc)) from exc
 
-    async def _delete_sandbox(self, sandbox_id: str) -> None:
-        headers = {"X-Inner-Api-Key": self.auth_token} if self.auth_token else {}
-        encoded_sandbox_id = quote(sandbox_id, safe="")
-        async with httpx.AsyncClient(
-            base_url=self.gateway_endpoint.rstrip("/"),
-            headers=headers,
-            timeout=httpx.Timeout(self.gateway_timeout),
-        ) as client:
-            response = await client.delete(f"/v1/sandboxes/{encoded_sandbox_id}")
-            if response.status_code == 404:
-                return
-            _ = response.raise_for_status()
-
-    async def _delete_sandbox_best_effort(self, sandbox_id: str) -> None:
-        try:
-            await self._delete_sandbox(sandbox_id)
-        except BaseException:
-            logger.warning(
-                "failed to delete Enterprise sandbox after Binding creation failed",
-                exc_info=True,
-                extra={"binding_ref": sandbox_id},
-            )
-
     async def _create_data_plane(self, binding_ref: str) -> ShellctlRuntimeLease:
         proxy_base_url = f"{self.gateway_endpoint.rstrip('/')}/proxy/"
         headers = {"X-Sandbox-Id": binding_ref}

dify-agent/src/dify_agent/runtime_backend/local.py

@@ -27,6 +27,7 @@ from dify_agent.runtime_backend.protocols import (
     ExecutionBindingCreateSpec,
     ExecutionBindingDestroySpec,
     HomeSnapshotCreateSpec,
+    InitializeHomeSnapshotSpec,
     RuntimeLayout,
     RuntimeLease,
 )
@@ -48,6 +49,28 @@ class LocalHomeSnapshotBackend:
     snapshot_root: str = "/home/dify/.dify-agent-home-snapshots"
     client_factory: ShellctlClientFactory | None = None
 
+    async def initialize(self, spec: InitializeHomeSnapshotSpec) -> str:
+        snapshot_ref = _local_snapshot_ref(spec.home_snapshot_id)
+        lease = self._control_lease(snapshot_ref)
+        target = self._snapshot_dir(snapshot_ref)
+        try:
+            result = await run_shellctl_control_command(
+                lease.commands,
+                f"set -eu\nmkdir -p {shlex.quote(target)}\nchmod 700 {shlex.quote(target)}",
+            )
+            if result.exit_code != 0:
+                raise HomeSnapshotCreateError(result.output)
+            return snapshot_ref
+        except BaseException as exc:
+            await _remove_partial(lease.commands, target=target, resource_ref=snapshot_ref)
+            if isinstance(exc, HomeSnapshotCreateError):
+                raise
+            if isinstance(exc, Exception):
+                raise HomeSnapshotCreateError(str(exc)) from exc
+            raise
+        finally:
+            await _close_best_effort(lease, resource_ref=snapshot_ref)
+
     async def create_from_runtime(self, *, spec: HomeSnapshotCreateSpec, source: RuntimeLease) -> str:
         snapshot_ref = _local_snapshot_ref(spec.home_snapshot_id)
         target = self._snapshot_dir(snapshot_ref)
@@ -119,31 +142,31 @@ class LocalExecutionBindingBackend:
     async def create_binding(self, spec: ExecutionBindingCreateSpec) -> ExecutionBindingAllocation:
         binding_id = _validated_ref_part(spec.binding_id)
         workspace_id = _validated_ref_part(spec.workspace_id)
+        snapshot_ref = _validated_ref_part(spec.home_snapshot_ref)
         workspace_ref = workspace_id
         if spec.existing_workspace_ref is not None:
             existing_workspace_ref = _validated_ref_part(spec.existing_workspace_ref)
             if existing_workspace_ref != workspace_ref:
                 raise BindingCreateError("existing Workspace ref does not match workspace_id")
-        snapshot_dir: str | None = None
-        if spec.home_snapshot_ref is not None:
-            snapshot_ref = _validated_ref_part(spec.home_snapshot_ref)
-            snapshot_dir = f"{self.snapshot_root.rstrip('/')}/{snapshot_ref}"
         binding_ref = _local_binding_ref(binding_id=binding_id, workspace_id=workspace_id)
+        lease = self._control_lease(binding_ref)
         home_dir = self._home_dir(binding_id)
         workspace_dir = self._workspace_dir(workspace_id)
+        snapshot_dir = f"{self.snapshot_root.rstrip('/')}/{snapshot_ref}"
         creates_workspace = spec.existing_workspace_ref is None
         workspace_setup = (
             f"mkdir -p {shlex.quote(workspace_dir)}" if creates_workspace else f"test -d {shlex.quote(workspace_dir)}"
         )
-        setup = ["set -eu"]
-        if snapshot_dir is not None:
-            setup.append(f"test -d {shlex.quote(snapshot_dir)}")
-        setup.extend([workspace_setup, f"mkdir -p {shlex.quote(home_dir)}"])
-        if snapshot_dir is not None:
-            setup.append(f"cp -a {shlex.quote(snapshot_dir)}/. {shlex.quote(home_dir)}/")
-        setup.append(f"chmod 700 {shlex.quote(home_dir)} {shlex.quote(workspace_dir)}")
-        script = "\n".join(setup)
-        lease = self._control_lease(binding_ref)
+        script = "\n".join(
+            [
+                "set -eu",
+                f"test -d {shlex.quote(snapshot_dir)}",
+                workspace_setup,
+                f"mkdir -p {shlex.quote(home_dir)}",
+                f"cp -a {shlex.quote(snapshot_dir)}/. {shlex.quote(home_dir)}/",
+                f"chmod 700 {shlex.quote(home_dir)} {shlex.quote(workspace_dir)}",
+            ]
+        )
         try:
             result = await run_shellctl_control_command(lease.commands, script)
             if result.exit_code != 0:

dify-agent/src/dify_agent/runtime_backend/profile.py

@@ -99,7 +99,11 @@ def create_runtime_backend_profile(settings: RuntimeBackendSettings) -> RuntimeB
             endpoint = settings.enterprise_sandbox_gateway_endpoint or ""
             token = settings.enterprise_sandbox_gateway_auth_token or ""
             return RuntimeBackendProfile(
-                home_snapshots=EnterpriseHomeSnapshotBackend(),
+                home_snapshots=EnterpriseHomeSnapshotBackend(
+                    gateway_endpoint=endpoint,
+                    auth_token=token,
+                    gateway_timeout=settings.enterprise_sandbox_gateway_timeout,
+                ),
                 execution_bindings=EnterpriseExecutionBindingBackend(
                     gateway_endpoint=endpoint,
                     auth_token=token,
@@ -112,10 +116,11 @@ def create_runtime_backend_profile(settings: RuntimeBackendSettings) -> RuntimeB
             return RuntimeBackendProfile(
                 home_snapshots=E2BHomeSnapshotBackend(
                     control_plane=control_plane,
+                    template=settings.e2b_template,
+                    active_timeout_seconds=settings.e2b_active_timeout_seconds,
                 ),
                 execution_bindings=E2BExecutionBindingBackend(
                     control_plane=control_plane,
-                    template=settings.e2b_template,
                     active_timeout_seconds=settings.e2b_active_timeout_seconds,
                     shellctl_auth_token=settings.e2b_shellctl_auth_token,
                     shellctl_port=settings.e2b_shellctl_port,

dify-agent/src/dify_agent/runtime_backend/protocols.py

@@ -13,6 +13,13 @@ from typing import Protocol
 from dify_agent.adapters.shell.protocols import ShellCommandProtocol
 
 
+@dataclass(frozen=True, slots=True)
+class InitializeHomeSnapshotSpec:
+    tenant_id: str
+    agent_id: str
+    home_snapshot_id: str
+
+
 @dataclass(frozen=True, slots=True)
 class HomeSnapshotCreateSpec:
     tenant_id: str
@@ -93,7 +100,7 @@ class ExecutionBindingCreateSpec:
     binding_id: str
     workspace_id: str
     existing_workspace_ref: str | None
-    home_snapshot_ref: str | None = None
+    home_snapshot_ref: str
 
 
 @dataclass(frozen=True, slots=True)
@@ -119,18 +126,12 @@ class ExecutionBindingBackend(Protocol):
     async def create_binding(self, spec: ExecutionBindingCreateSpec) -> ExecutionBindingAllocation:
         """Materialize a mutable Home and make the requested Workspace ready.
 
-        With a non-null ``home_snapshot_ref``, implementations must initialize
-        the Home from that exact immutable snapshot and must fail rather than
-        fall back when it is unavailable. With ``None``, implementations must
-        create an independent mutable Home from their deployment default without
-        implicitly creating an immutable snapshot.
-
-        Implementations must return stable opaque refs only after Home and
-        Workspace are usable. With an ``existing_workspace_ref``, they must
-        attach that Workspace without clearing or replacing its contents;
-        unsupported sharing must fail before mutating it. On failure,
-        implementations should clean up newly allocated partial resources and
-        must not damage a pre-existing Workspace.
+        Implementations must initialize the Home from ``home_snapshot_ref`` and
+        return stable opaque refs only after both resources are usable. With an
+        ``existing_workspace_ref``, they must attach that Workspace without
+        clearing or replacing its contents; unsupported sharing must fail before
+        mutating it. On failure, implementations should clean up newly allocated
+        partial resources and must not damage a pre-existing Workspace.
         """
         ...
 
@@ -170,6 +171,16 @@ class ExecutionBindingBackend(Protocol):
 class HomeSnapshotBackend(Protocol):
     """Manage immutable backend-native Home resources."""
 
+    async def initialize(self, spec: InitializeHomeSnapshotSpec) -> str:
+        """Create the deployment-defined baseline Home Snapshot.
+
+        Implementations may use any backend-native bootstrap mechanism, but must
+        return a stable opaque ref only after an immutable snapshot is ready for
+        future Binding creation. Temporary bootstrap resources must not become
+        part of the logical snapshot lifecycle.
+        """
+        ...
+
     async def create_from_runtime(self, *, spec: HomeSnapshotCreateSpec, source: RuntimeLease) -> str:
         """Capture the source lease's current Home as a new immutable snapshot.
 
@@ -206,6 +217,7 @@ __all__ = [
     "FileSystem",
     "HomeSnapshotBackend",
     "HomeSnapshotCreateSpec",
+    "InitializeHomeSnapshotSpec",
     "RuntimeBackendProfile",
     "RuntimeLayout",
     "RuntimeLease",

dify-agent/src/dify_agent/server/home_snapshots.py

@@ -6,6 +6,7 @@ from dify_agent.protocol.home_snapshot import (
     CreateHomeSnapshotFromBindingRequest,
     DeleteHomeSnapshotRequest,
     HomeSnapshotResponse,
+    InitializeHomeSnapshotRequest,
 )
 from dify_agent.runtime_backend import (
     BindingAcquireError,
@@ -15,6 +16,7 @@ from dify_agent.runtime_backend import (
     HomeSnapshotCreateError,
     HomeSnapshotCreateSpec,
     HomeSnapshotNotFoundError,
+    InitializeHomeSnapshotSpec,
 )
 from dify_agent.runtime_backend.leases import open_runtime_lease
 
@@ -36,6 +38,19 @@ class HomeSnapshotService:
     home_snapshots: HomeSnapshotBackend
     execution_bindings: ExecutionBindingBackend
 
+    async def initialize(self, request: InitializeHomeSnapshotRequest) -> HomeSnapshotResponse:
+        try:
+            snapshot_ref = await self.home_snapshots.initialize(
+                InitializeHomeSnapshotSpec(
+                    tenant_id=request.tenant_id,
+                    agent_id=request.agent_id,
+                    home_snapshot_id=request.home_snapshot_id,
+                )
+            )
+        except HomeSnapshotCreateError as exc:
+            raise HomeSnapshotServiceError("home_snapshot_create_failed", str(exc), status_code=502) from exc
+        return HomeSnapshotResponse(snapshot_ref=snapshot_ref)
+
     async def create_from_binding(
         self,
         request: CreateHomeSnapshotFromBindingRequest,

dify-agent/src/dify_agent/server/routes/home_snapshots.py

@@ -9,6 +9,7 @@ from dify_agent.protocol.home_snapshot import (
     CreateHomeSnapshotFromBindingRequest,
     DeleteHomeSnapshotRequest,
     HomeSnapshotResponse,
+    InitializeHomeSnapshotRequest,
 )
 from dify_agent.server.home_snapshots import HomeSnapshotService, HomeSnapshotServiceError
 
@@ -25,6 +26,19 @@ def create_home_snapshots_router(get_service: Callable[[], HomeSnapshotService |
             )
         return service
 
+    @router.post("/initialize", response_model=HomeSnapshotResponse, status_code=status.HTTP_201_CREATED)
+    async def initialize_snapshot(
+        request: InitializeHomeSnapshotRequest,
+        service: Annotated[HomeSnapshotService, Depends(service_dep)],
+    ) -> HomeSnapshotResponse:
+        try:
+            return await service.initialize(request)
+        except HomeSnapshotServiceError as exc:
+            raise HTTPException(
+                status_code=exc.status_code,
+                detail={"code": exc.code, "message": exc.message},
+            ) from exc
+
     @router.post("/from-binding", response_model=HomeSnapshotResponse, status_code=status.HTTP_201_CREATED)
     async def create_snapshot_from_binding(
         request: CreateHomeSnapshotFromBindingRequest,

dify-agent/tests/integration/dify_agent/runtime_backend/test_working_environment.py

@@ -12,9 +12,10 @@ from dify_agent.runtime_backend import (
     ExecutionBindingCreateSpec,
     ExecutionBindingDestroySpec,
     HomeSnapshotCreateSpec,
+    InitializeHomeSnapshotSpec,
 )
 from dify_agent.runtime_backend.e2b import E2BExecutionBindingBackend, E2BHomeSnapshotBackend, E2BSDKControlPlane
-from dify_agent.runtime_backend.local import LocalExecutionBindingBackend
+from dify_agent.runtime_backend.local import LocalExecutionBindingBackend, LocalHomeSnapshotBackend
 
 pytestmark = pytest.mark.integration
 
@@ -31,10 +32,19 @@ async def test_local_two_agents_share_workspace_but_not_home() -> None:
     endpoint = _required_env("DIFY_AGENT_TEST_LOCAL_SHELLCTL_ENDPOINT", "real Local shellctl")
     token = os.environ.get("DIFY_AGENT_TEST_LOCAL_SHELLCTL_AUTH_TOKEN", "")
     marker = uuid.uuid4().hex
+    snapshots = LocalHomeSnapshotBackend(endpoint=endpoint, auth_token=token)
     bindings = LocalExecutionBindingBackend(endpoint=endpoint, auth_token=token)
+    snapshot_ref: str | None = None
     allocations = []
     active_leases = []
     try:
+        snapshot_ref = await snapshots.initialize(
+            InitializeHomeSnapshotSpec(
+                tenant_id="integration-tenant",
+                agent_id="integration-agent",
+                home_snapshot_id=marker,
+            )
+        )
         first = await bindings.create_binding(
             ExecutionBindingCreateSpec(
                 tenant_id="integration-tenant",
@@ -42,7 +52,7 @@ async def test_local_two_agents_share_workspace_but_not_home() -> None:
                 binding_id=f"binding-a-{marker}",
                 workspace_id=f"workspace-{marker}",
                 existing_workspace_ref=None,
-                home_snapshot_ref=None,
+                home_snapshot_ref=snapshot_ref,
             )
         )
         allocations.append(first)
@@ -61,7 +71,7 @@ async def test_local_two_agents_share_workspace_but_not_home() -> None:
                 binding_id=f"binding-b-{marker}",
                 workspace_id=f"workspace-{marker}",
                 existing_workspace_ref=first.workspace_ref,
-                home_snapshot_ref=None,
+                home_snapshot_ref=snapshot_ref,
             )
         )
         allocations.append(second)
@@ -92,6 +102,11 @@ async def test_local_two_agents_share_workspace_but_not_home() -> None:
                 )
             except BaseException as exc:
                 cleanup_errors.append(exc)
+        if snapshot_ref is not None:
+            try:
+                await snapshots.delete(snapshot_ref)
+            except BaseException as exc:
+                cleanup_errors.append(exc)
         if cleanup_errors and not primary_error:
             raise cleanup_errors[0]
 
@@ -105,14 +120,22 @@ async def test_e2b_binding_checkpoint_and_collection() -> None:
     )
     marker = uuid.uuid4().hex
     control = E2BSDKControlPlane(api_key=api_key)
-    snapshots = E2BHomeSnapshotBackend(control_plane=control)
-    bindings = E2BExecutionBindingBackend(control_plane=control, template=template, active_timeout_seconds=3600)
+    snapshots = E2BHomeSnapshotBackend(control_plane=control, template=template, active_timeout_seconds=3600)
+    bindings = E2BExecutionBindingBackend(control_plane=control, active_timeout_seconds=3600)
+    snapshot_ref: str | None = None
     checkpoint_ref: str | None = None
     allocation = None
     checkpoint_allocation = None
     lease = None
     checkpoint_lease = None
     try:
+        snapshot_ref = await snapshots.initialize(
+            InitializeHomeSnapshotSpec(
+                tenant_id="integration-tenant",
+                agent_id="integration-agent",
+                home_snapshot_id=marker,
+            )
+        )
         allocation = await bindings.create_binding(
             ExecutionBindingCreateSpec(
                 tenant_id="integration-tenant",
@@ -120,7 +143,7 @@ async def test_e2b_binding_checkpoint_and_collection() -> None:
                 binding_id=marker,
                 workspace_id=marker,
                 existing_workspace_ref=None,
-                home_snapshot_ref=None,
+                home_snapshot_ref=snapshot_ref,
             )
         )
         lease = await bindings.acquire(allocation.binding_ref)
@@ -188,10 +211,11 @@ async def test_e2b_binding_checkpoint_and_collection() -> None:
                 )
             except BaseException as exc:
                 cleanup_errors.append(exc)
-        if checkpoint_ref is not None:
-            try:
-                await snapshots.delete(checkpoint_ref)
-            except BaseException as exc:
-                cleanup_errors.append(exc)
+        for ref in (checkpoint_ref, snapshot_ref):
+            if ref is not None:
+                try:
+                    await snapshots.delete(ref)
+                except BaseException as exc:
+                    cleanup_errors.append(exc)
         if cleanup_errors and not primary_error:
             raise cleanup_errors[0]

dify-agent/tests/local/dify_agent/client/test_client.py

@@ -28,6 +28,7 @@ from dify_agent.protocol import (
     CreateHomeSnapshotFromBindingRequest,
     CreateRunRequest,
     DestroyExecutionBindingRequest,
+    InitializeHomeSnapshotRequest,
     RUN_EVENT_ADAPTER,
     RunCancelledEvent,
     RunEvent,
@@ -310,6 +311,14 @@ def test_async_workspace_methods_post_dtos_and_parse_responses() -> None:
     asyncio.run(scenario())
 
 
+def _initialize_home_snapshot_request() -> InitializeHomeSnapshotRequest:
+    return InitializeHomeSnapshotRequest(
+        tenant_id="tenant-1",
+        agent_id="agent-1",
+        home_snapshot_id="home-1",
+    )
+
+
 def test_sync_execution_binding_client_uses_private_binding_routes() -> None:
     def handler(request: httpx.Request) -> httpx.Response:
         payload = cast(dict[str, object], json.loads(request.content))
@@ -359,9 +368,12 @@ def _create_home_snapshot_from_binding_request() -> CreateHomeSnapshotFromBindin
     )
 
 
-def test_sync_home_snapshot_client_parses_checkpoint_and_delete() -> None:
+def test_sync_home_snapshot_client_parses_initialize_checkpoint_and_delete() -> None:
     def handler(request: httpx.Request) -> httpx.Response:
         if request.method == "POST":
+            if request.url.path == "/home-snapshots/initialize":
+                assert json.loads(request.content) == _initialize_home_snapshot_request().model_dump(mode="json")
+                return httpx.Response(201, json={"snapshot_ref": "initial-home"})
             if request.url.path == "/home-snapshots/from-binding":
                 assert json.loads(request.content) == _create_home_snapshot_from_binding_request().model_dump(
                     mode="json"
@@ -375,16 +387,20 @@ def test_sync_home_snapshot_client_parses_checkpoint_and_delete() -> None:
     http_client = httpx.Client(transport=httpx.MockTransport(handler))
     client = Client(base_url="http://testserver", sync_http_client=http_client)
 
+    initialized = client.initialize_home_snapshot_sync(_initialize_home_snapshot_request())
     created = client.create_home_snapshot_from_binding_sync(_create_home_snapshot_from_binding_request())
     client.delete_home_snapshot_sync(created.snapshot_ref)
 
+    assert initialized.snapshot_ref == "initial-home"
     assert created.snapshot_ref == "team/home 1"
     http_client.close()
 
 
-def test_async_home_snapshot_client_parses_checkpoint_and_delete() -> None:
+def test_async_home_snapshot_client_parses_initialize_checkpoint_and_delete() -> None:
     def handler(request: httpx.Request) -> httpx.Response:
         if request.method == "POST":
+            if request.url.path == "/home-snapshots/initialize":
+                return httpx.Response(201, json={"snapshot_ref": "initial-home"})
             if request.url.path == "/home-snapshots/from-binding":
                 return httpx.Response(201, json={"snapshot_ref": "team/home 1"})
             assert request.url.path == "/home-snapshots/delete"
@@ -395,9 +411,11 @@ def test_async_home_snapshot_client_parses_checkpoint_and_delete() -> None:
         http_client = httpx.AsyncClient(transport=httpx.MockTransport(handler))
         client = Client(base_url="http://testserver", async_http_client=http_client)
 
+        initialized = await client.initialize_home_snapshot(_initialize_home_snapshot_request())
         created = await client.create_home_snapshot_from_binding(_create_home_snapshot_from_binding_request())
         await client.delete_home_snapshot(created.snapshot_ref)
 
+        assert initialized.snapshot_ref == "initial-home"
         assert created.snapshot_ref == "team/home 1"
         await http_client.aclose()
 
@@ -412,7 +430,7 @@ def test_home_snapshot_client_maps_sync_validation_and_async_http_errors() -> No
     )
 
     with pytest.raises(DifyAgentValidationError):
-        _ = sync_client.create_home_snapshot_from_binding_sync(_create_home_snapshot_from_binding_request())
+        _ = sync_client.initialize_home_snapshot_sync(_initialize_home_snapshot_request())
     sync_http_client.close()
 
     async def scenario() -> None:

dify-agent/tests/local/dify_agent/protocol/test_working_environment.py

@@ -29,29 +29,6 @@ def test_execution_binding_request_uses_opaque_backend_refs() -> None:
     }
 
 
-def test_execution_binding_request_accepts_missing_or_null_home_snapshot_ref() -> None:
-    fields = {
-        "tenant_id": "tenant-1",
-        "agent_id": "agent-1",
-        "binding_id": "binding-1",
-        "workspace_id": "workspace-1",
-    }
-
-    assert CreateExecutionBindingRequest(**fields).home_snapshot_ref is None
-    assert CreateExecutionBindingRequest(**fields, home_snapshot_ref=None).home_snapshot_ref is None
-
-
-def test_execution_binding_request_rejects_empty_home_snapshot_ref() -> None:
-    with pytest.raises(ValidationError, match="home_snapshot_ref"):
-        CreateExecutionBindingRequest(
-            tenant_id="tenant-1",
-            agent_id="agent-1",
-            binding_id="binding-1",
-            workspace_id="workspace-1",
-            home_snapshot_ref="",
-        )
-
-
 def test_destroy_workspace_requires_workspace_ref() -> None:
     with pytest.raises(ValidationError, match="workspace_ref"):
         DestroyExecutionBindingRequest(binding_ref="binding-1", destroy_workspace=True)

dify-agent/tests/local/dify_agent/runtime_backend/test_e2b.py

@@ -10,6 +10,7 @@ from dify_agent.runtime_backend import (
     ExecutionBindingCreateSpec,
     ExecutionBindingDestroySpec,
     HomeSnapshotCreateSpec,
+    InitializeHomeSnapshotSpec,
     SharedWorkspaceUnsupportedError,
     WorkspacePreservationUnsupportedError,
 )
@@ -85,7 +86,7 @@ class _ControlPlane:
         sandbox = _Sandbox(sandbox_id=sandbox_id, pause_error=self.pause_error)
         self.sandboxes[sandbox_id] = sandbox
         self.created.append((template, on_timeout))
-        assert metadata["dify.resource"] == "runtime-sandbox"
+        assert metadata["dify.resource"] in {"home-snapshot-initialize", "runtime-sandbox"}
         return sandbox
 
     async def connect(self, handle: str, *, timeout: int) -> _Sandbox:
@@ -102,57 +103,49 @@ class _ControlPlane:
 
 
 @pytest.mark.anyio
-async def test_e2b_binding_uses_default_template_or_exact_snapshot_and_couples_refs() -> None:
+async def test_e2b_profile_uses_snapshot_as_runtime_template_and_couples_refs() -> None:
     control = _ControlPlane()
     snapshots = E2BHomeSnapshotBackend(
         control_plane=control,  # pyright: ignore[reportArgumentType]
-    )
-    bindings = E2BExecutionBindingBackend(
-        control_plane=control,  # pyright: ignore[reportArgumentType]
         template="prepared-template",
         active_timeout_seconds=3600,
     )
+    bindings = E2BExecutionBindingBackend(
+        control_plane=control,  # pyright: ignore[reportArgumentType]
+        active_timeout_seconds=3600,
+    )
 
-    default_allocation = await bindings.create_binding(
+    snapshot_ref = await snapshots.initialize(
+        InitializeHomeSnapshotSpec(tenant_id="tenant-1", agent_id="agent-1", home_snapshot_id="home-1")
+    )
+    allocation = await bindings.create_binding(
         ExecutionBindingCreateSpec(
             tenant_id="tenant-1",
             agent_id="agent-1",
             binding_id="binding-1",
             workspace_id="workspace-1",
             existing_workspace_ref=None,
-            home_snapshot_ref=None,
-        )
-    )
-    snapshot_allocation = await bindings.create_binding(
-        ExecutionBindingCreateSpec(
-            tenant_id="tenant-1",
-            agent_id="agent-1",
-            binding_id="binding-2",
-            workspace_id="workspace-2",
-            existing_workspace_ref=None,
-            home_snapshot_ref="snapshot-1",
+            home_snapshot_ref=snapshot_ref,
         )
     )
 
-    assert control.created == [("prepared-template", "pause"), ("snapshot-1", "pause")]
-    assert default_allocation.binding_ref == default_allocation.workspace_ref
-    assert snapshot_allocation.binding_ref == snapshot_allocation.workspace_ref
-    runtime = control.sandboxes[default_allocation.binding_ref]
+    assert control.created == [("prepared-template", "kill"), (snapshot_ref, "pause")]
+    assert allocation.binding_ref == allocation.workspace_ref
+    runtime = control.sandboxes[allocation.binding_ref]
     assert runtime.files.paths == {"/home/dify/workspace"}
     assert runtime.pauses == [True]
 
-    for allocation in (default_allocation, snapshot_allocation):
-        await bindings.destroy_binding(
-            ExecutionBindingDestroySpec(
-                binding_ref=allocation.binding_ref,
-                workspace_ref=allocation.workspace_ref,
-                destroy_workspace=True,
-            )
+    await bindings.destroy_binding(
+        ExecutionBindingDestroySpec(
+            binding_ref=allocation.binding_ref,
+            workspace_ref=allocation.workspace_ref,
+            destroy_workspace=True,
         )
-    await snapshots.delete("snapshot-1")
+    )
+    await snapshots.delete(snapshot_ref)
 
-    assert control.killed == [default_allocation.binding_ref, snapshot_allocation.binding_ref]
-    assert control.deleted_snapshots == ["snapshot-1"]
+    assert control.killed == [allocation.binding_ref]
+    assert control.deleted_snapshots == [snapshot_ref]
 
 
 @pytest.mark.anyio
@@ -160,7 +153,6 @@ async def test_e2b_rejects_shared_workspace_and_binding_only_destroy() -> None:
     control = _ControlPlane()
     backend = E2BExecutionBindingBackend(
         control_plane=control,  # pyright: ignore[reportArgumentType]
-        template="prepared-template",
         active_timeout_seconds=3600,
     )
     spec = ExecutionBindingCreateSpec(
@@ -185,7 +177,6 @@ async def test_e2b_binding_create_kills_sandbox_when_initialization_fails() -> N
     control = _ControlPlane(pause_error=RuntimeError("pause failed"))
     backend = E2BExecutionBindingBackend(
         control_plane=control,  # pyright: ignore[reportArgumentType]
-        template="prepared-template",
         active_timeout_seconds=3600,
     )
 
@@ -205,36 +196,6 @@ async def test_e2b_binding_create_kills_sandbox_when_initialization_fails() -> N
     assert sandbox.killed == 1
 
 
-@pytest.mark.anyio
-async def test_e2b_missing_explicit_snapshot_does_not_fall_back_to_template() -> None:
-    class _FailingControlPlane(_ControlPlane):
-        async def create(self, template: str, *, timeout: int, metadata: dict[str, str], on_timeout: str) -> _Sandbox:
-            del timeout, metadata
-            self.created.append((template, on_timeout))
-            raise RuntimeError("snapshot unavailable")
-
-    control = _FailingControlPlane()
-    backend = E2BExecutionBindingBackend(
-        control_plane=control,  # pyright: ignore[reportArgumentType]
-        template="prepared-template",
-        active_timeout_seconds=3600,
-    )
-
-    with pytest.raises(BindingCreateError, match="snapshot unavailable"):
-        await backend.create_binding(
-            ExecutionBindingCreateSpec(
-                tenant_id="tenant-1",
-                agent_id="agent-1",
-                binding_id="binding-1",
-                workspace_id="workspace-1",
-                existing_workspace_ref=None,
-                home_snapshot_ref="missing-snapshot",
-            )
-        )
-
-    assert control.created == [("missing-snapshot", "pause")]
-
-
 @pytest.mark.anyio
 async def test_e2b_checkpoint_uses_exact_source_runtime() -> None:
     control = _ControlPlane()
@@ -245,6 +206,8 @@ async def test_e2b_checkpoint_uses_exact_source_runtime() -> None:
     )
     backend = E2BHomeSnapshotBackend(
         control_plane=control,  # pyright: ignore[reportArgumentType]
+        template="prepared-template",
+        active_timeout_seconds=3600,
     )
 
     snapshot_ref = await backend.create_from_runtime(

dify-agent/tests/local/dify_agent/runtime_backend/test_enterprise_backend.py

@@ -9,14 +9,13 @@ import pytest
 
 from dify_agent.runtime_backend import (
     BindingAcquireError,
-    BindingCreateError,
     BindingDestroyError,
     BindingLostError,
     ExecutionBindingCreateSpec,
     ExecutionBindingDestroySpec,
     HomeSnapshotCreateSpec,
+    InitializeHomeSnapshotSpec,
     RuntimeLease,
-    SharedWorkspaceUnsupportedError,
     WorkspacePreservationUnsupportedError,
 )
 from dify_agent.runtime_backend.enterprise import (
@@ -283,55 +282,23 @@ async def test_enterprise_destroy_propagates_gateway_failure(
 
 
 @pytest.mark.anyio
-async def test_enterprise_default_binding_creates_gateway_sandbox_and_layout(
-    monkeypatch: pytest.MonkeyPatch,
-) -> None:
-    requests: list[httpx.Request] = []
+async def test_enterprise_allocation_and_home_snapshots_remain_explicitly_not_implemented() -> None:
+    snapshots = EnterpriseHomeSnapshotBackend(gateway_endpoint="https://gateway", auth_token="secret")
+    bindings = EnterpriseExecutionBindingBackend(gateway_endpoint="https://gateway", auth_token="secret")
 
-    def handler(request: httpx.Request) -> httpx.Response:
-        requests.append(request)
-        if request.url.path == "/v1/sandboxes":
-            assert json.loads(request.content) == {"tenantId": "tenant-1"}
-            return httpx.Response(201, json={"sandboxId": "sandbox-1", "status": "running"})
-        if request.url.path == "/proxy/v1/jobs/run":
-            assert request.headers["X-Sandbox-Id"] == "sandbox-1"
-            payload = cast(dict[str, object], json.loads(request.content))
-            script = payload["script"]
-            assert isinstance(script, str)
-            assert "mkdir -p /home/dify" in script
-            assert "rm -rf -- /home/dify/workspace" in script
-            return _job_response()
-        return httpx.Response(200, json={"job_id": "job-1"})
-
-    clients = _mock_http(monkeypatch, handler)
-    backend = EnterpriseExecutionBindingBackend(gateway_endpoint="http://gateway.example", auth_token="secret")
-
-    allocation = await backend.create_binding(
-        ExecutionBindingCreateSpec(
-            tenant_id="tenant-1",
-            agent_id="agent-1",
-            binding_id="binding-1",
-            workspace_id="workspace-1",
-            existing_workspace_ref=None,
-            home_snapshot_ref=None,
+    with pytest.raises(NotImplementedError, match="Execution Binding protocol"):
+        _ = await snapshots.initialize(
+            InitializeHomeSnapshotSpec(tenant_id="tenant-1", agent_id="agent-1", home_snapshot_id="home-1")
         )
-    )
-
-    assert allocation.binding_ref == allocation.workspace_ref == "sandbox-1"
-    assert requests[0].headers["X-Inner-Api-Key"] == "secret"
-    assert all(client.is_closed for client in clients)
-
-
-@pytest.mark.anyio
-async def test_enterprise_binding_rejects_snapshot_and_shared_workspace_before_gateway_call(
-    monkeypatch: pytest.MonkeyPatch,
-) -> None:
-    requests: list[httpx.Request] = []
-    _ = _mock_http(monkeypatch, lambda request: requests.append(request) or httpx.Response(500))
-    backend = EnterpriseExecutionBindingBackend(gateway_endpoint="http://gateway.example", auth_token="secret")
-
-    with pytest.raises(BindingCreateError, match="immutable Home Snapshot"):
-        await backend.create_binding(
+    with pytest.raises(NotImplementedError, match="Execution Binding protocol"):
+        _ = await snapshots.create_from_runtime(
+            spec=HomeSnapshotCreateSpec(tenant_id="tenant-1", agent_id="agent-1", home_snapshot_id="home-2"),
+            source=cast(RuntimeLease, object()),
+        )
+    with pytest.raises(NotImplementedError, match="Execution Binding protocol"):
+        await snapshots.delete("snapshot-1")
+    with pytest.raises(NotImplementedError, match="Execution Binding protocol"):
+        _ = await bindings.create_binding(
             ExecutionBindingCreateSpec(
                 tenant_id="tenant-1",
                 agent_id="agent-1",
@@ -341,63 +308,3 @@ async def test_enterprise_binding_rejects_snapshot_and_shared_workspace_before_g
                 home_snapshot_ref="snapshot-1",
             )
         )
-    with pytest.raises(SharedWorkspaceUnsupportedError):
-        await backend.create_binding(
-            ExecutionBindingCreateSpec(
-                tenant_id="tenant-1",
-                agent_id="agent-1",
-                binding_id="binding-1",
-                workspace_id="workspace-1",
-                existing_workspace_ref="workspace-1",
-                home_snapshot_ref=None,
-            )
-        )
-
-    assert requests == []
-
-
-@pytest.mark.anyio
-async def test_enterprise_binding_create_deletes_new_sandbox_when_layout_setup_fails(
-    monkeypatch: pytest.MonkeyPatch,
-) -> None:
-    requests: list[httpx.Request] = []
-
-    def handler(request: httpx.Request) -> httpx.Response:
-        requests.append(request)
-        if request.url.path == "/v1/sandboxes":
-            return httpx.Response(201, json={"sandboxId": "sandbox-1"})
-        if request.url.path == "/proxy/v1/jobs/run":
-            return _job_response(exit_code=1)
-        if request.method == "DELETE":
-            return httpx.Response(204)
-        return httpx.Response(200, json={"job_id": "job-1"})
-
-    _ = _mock_http(monkeypatch, handler)
-    backend = EnterpriseExecutionBindingBackend(gateway_endpoint="http://gateway.example", auth_token="secret")
-
-    with pytest.raises(BindingCreateError):
-        await backend.create_binding(
-            ExecutionBindingCreateSpec(
-                tenant_id="tenant-1",
-                agent_id="agent-1",
-                binding_id="binding-1",
-                workspace_id="workspace-1",
-                existing_workspace_ref=None,
-                home_snapshot_ref=None,
-            )
-        )
-
-    assert any(request.method == "DELETE" and request.url.path == "/v1/sandboxes/sandbox-1" for request in requests)
-
-
-@pytest.mark.anyio
-async def test_enterprise_home_snapshots_remain_explicitly_not_implemented() -> None:
-    snapshots = EnterpriseHomeSnapshotBackend()
-
-    with pytest.raises(NotImplementedError, match="immutable Home Snapshot"):
-        _ = await snapshots.create_from_runtime(
-            spec=HomeSnapshotCreateSpec(tenant_id="tenant-1", agent_id="agent-1", home_snapshot_id="home-2"),
-            source=cast(RuntimeLease, object()),
-        )
-    with pytest.raises(NotImplementedError, match="immutable Home Snapshot"):
-        await snapshots.delete("snapshot-1")

dify-agent/tests/local/dify_agent/runtime_backend/test_local.py

@@ -13,6 +13,8 @@ from dify_agent.runtime_backend import (
     ExecutionBindingCreateSpec,
     ExecutionBindingDestroySpec,
     HomeSnapshotCreateSpec,
+    HomeSnapshotCreateError,
+    InitializeHomeSnapshotSpec,
 )
 from dify_agent.runtime_backend.local import LocalExecutionBindingBackend, LocalHomeSnapshotBackend
 
@@ -135,6 +137,42 @@ class _FailThenSucceedFactory:
         return tuple(command for run in self.runs for command in run.commands)
 
 
+@pytest.mark.anyio
+async def test_local_snapshot_initialize_creates_private_snapshot_directory() -> None:
+    factory = _Factory()
+    snapshots = LocalHomeSnapshotBackend(
+        endpoint="http://shellctl",
+        auth_token="",
+        snapshot_root="/snapshots",
+        client_factory=factory,  # pyright: ignore[reportArgumentType]
+    )
+    snapshot_ref = await snapshots.initialize(
+        InitializeHomeSnapshotSpec(tenant_id="tenant-1", agent_id="agent-1", home_snapshot_id="home-1")
+    )
+
+    assert snapshot_ref == "home-home-1"
+    assert ("mkdir", "-p", "/snapshots/home-home-1") in factory.commands
+    assert ("chmod", "700", "/snapshots/home-home-1") in factory.commands
+
+
+@pytest.mark.anyio
+async def test_local_snapshot_create_failure_removes_partial_snapshot() -> None:
+    factory = _FailThenSucceedFactory()
+    snapshots = LocalHomeSnapshotBackend(
+        endpoint="http://shellctl",
+        auth_token="",
+        snapshot_root="/snapshots",
+        client_factory=factory,  # pyright: ignore[reportArgumentType]
+    )
+
+    with pytest.raises(HomeSnapshotCreateError, match="primary shellctl failure"):
+        await snapshots.initialize(
+            InitializeHomeSnapshotSpec(tenant_id="tenant-1", agent_id="agent-1", home_snapshot_id="home-1")
+        )
+
+    assert ("rm", "-rf", "--", "/snapshots/home-home-1") in factory.commands
+
+
 @pytest.mark.anyio
 async def test_local_binding_create_materializes_home_and_new_workspace() -> None:
     factory = _Factory()
@@ -160,41 +198,13 @@ async def test_local_binding_create_materializes_home_and_new_workspace() -> Non
 
     assert allocation.binding_ref == "binding-1:workspace-1"
     assert allocation.workspace_ref == "workspace-1"
-    assert factory.commands[0] == ("test", "-d", "/snapshots/home-home-1")
+    assert ("test", "-d", "/snapshots/home-home-1") in factory.commands
     assert ("mkdir", "-p", "/workspaces/workspace-1") in factory.commands
     assert ("mkdir", "-p", "/homes/binding-1") in factory.commands
     assert ("cp", "-a", "/snapshots/home-home-1/.", "/homes/binding-1/") in factory.commands
     assert ("chmod", "700", "/homes/binding-1", "/workspaces/workspace-1") in factory.commands
 
 
-@pytest.mark.anyio
-async def test_local_binding_create_uses_empty_default_home_without_snapshot_access() -> None:
-    factory = _Factory()
-    backend = LocalExecutionBindingBackend(
-        endpoint="http://shellctl",
-        auth_token="",
-        materialized_home_root="/homes",
-        workspace_root="/workspaces",
-        snapshot_root="/snapshots",
-        client_factory=factory,  # pyright: ignore[reportArgumentType]
-    )
-
-    allocation = await backend.create_binding(
-        ExecutionBindingCreateSpec(
-            tenant_id="tenant-1",
-            agent_id="agent-1",
-            binding_id="binding-1",
-            workspace_id="workspace-1",
-            existing_workspace_ref=None,
-            home_snapshot_ref=None,
-        )
-    )
-
-    assert allocation.binding_ref == "binding-1:workspace-1"
-    assert ("mkdir", "-p", "/homes/binding-1") in factory.commands
-    assert all("/snapshots" not in part for command in factory.commands for part in command)
-
-
 @pytest.mark.anyio
 async def test_local_binding_create_failure_removes_partial_home_and_workspace() -> None:
     factory = _FailThenSucceedFactory()

dify-agent/tests/local/dify_agent/server/test_home_snapshots.py

@@ -8,16 +8,22 @@ import pytest
 from dify_agent.protocol import (
     CreateHomeSnapshotFromBindingRequest,
     DeleteHomeSnapshotRequest,
+    InitializeHomeSnapshotRequest,
 )
-from dify_agent.runtime_backend import HomeSnapshotCreateSpec, RuntimeLease
+from dify_agent.runtime_backend import HomeSnapshotCreateSpec, InitializeHomeSnapshotSpec, RuntimeLease
 from dify_agent.server.home_snapshots import HomeSnapshotService
 
 
 @dataclass(slots=True)
 class _HomeBackend:
+    initialized: list[InitializeHomeSnapshotSpec] = field(default_factory=list)
     checkpointed: list[tuple[HomeSnapshotCreateSpec, RuntimeLease]] = field(default_factory=list)
     deleted: list[str] = field(default_factory=list)
 
+    async def initialize(self, spec: InitializeHomeSnapshotSpec) -> str:
+        self.initialized.append(spec)
+        return "snapshot-initial"
+
     async def create_from_runtime(self, *, spec: HomeSnapshotCreateSpec, source: RuntimeLease) -> str:
         self.checkpointed.append((spec, source))
         return "snapshot-build"
@@ -41,7 +47,7 @@ class _BindingBackend:
 
 
 @pytest.mark.anyio
-async def test_home_snapshot_service_checkpoints_exact_binding() -> None:
+async def test_home_snapshot_service_initializes_and_checkpoints_exact_binding() -> None:
     lease = cast(RuntimeLease, object())
     homes = _HomeBackend()
     bindings = _BindingBackend(lease=lease)
@@ -50,6 +56,9 @@ async def test_home_snapshot_service_checkpoints_exact_binding() -> None:
         execution_bindings=bindings,  # pyright: ignore[reportArgumentType]
     )
 
+    initial = await service.initialize(
+        InitializeHomeSnapshotRequest(tenant_id="tenant-1", agent_id="agent-1", home_snapshot_id="home-1")
+    )
     checkpoint = await service.create_from_binding(
         CreateHomeSnapshotFromBindingRequest(
             tenant_id="tenant-1",
@@ -59,6 +68,7 @@ async def test_home_snapshot_service_checkpoints_exact_binding() -> None:
         )
     )
 
+    assert initial.snapshot_ref == "snapshot-initial"
     assert checkpoint.snapshot_ref == "snapshot-build"
     assert bindings.acquired == ["binding-ref"]
     assert bindings.released == [lease]

dify-agent/tests/local/dify_agent/server/test_settings.py

@@ -301,7 +301,6 @@ def test_build_runtime_backend_profile_passes_e2b_active_timeout() -> None:
     assert profile is not None
     assert isinstance(profile.execution_bindings, E2BExecutionBindingBackend)
     assert profile.execution_bindings.active_timeout_seconds == 900
-    assert profile.execution_bindings.template == "difys-default-team/dify-agent-local-sandbox"
 
 
 def test_sandbox_file_upload_limit_defaults_to_tool_file_limit() -> None:

docker/docker-compose.yaml

@@ -21,50 +21,12 @@ x-shared-api-worker-config: &shared-api-worker-config
       required: false
     - path: ./envs/vectorstores/weaviate.env
       required: false
-    - path: ./envs/vectorstores/qdrant.env
-      required: false
-    - path: ./envs/vectorstores/oceanbase.env
-      required: false
-    - path: ./envs/vectorstores/seekdb.env
-      required: false
-    - path: ./envs/vectorstores/couchbase.env
-      required: false
-    - path: ./envs/vectorstores/pgvector.env
-      required: false
-    - path: ./envs/vectorstores/vastbase.env
-      required: false
-    - path: ./envs/vectorstores/pgvecto-rs.env
-      required: false
-    - path: ./envs/vectorstores/chroma.env
-      required: false
-    - path: ./envs/vectorstores/iris.env
-      required: false
-    - path: ./envs/vectorstores/oracle.env
-      required: false
-    - path: ./envs/vectorstores/opengauss.env
-      required: false
-    - path: ./envs/vectorstores/myscale.env
-      required: false
-    - path: ./envs/vectorstores/matrixone.env
-      required: false
-    - path: ./envs/vectorstores/elasticsearch.env
-      required: false
-    - path: ./envs/vectorstores/opensearch.env
-      required: false
-    - path: ./envs/vectorstores/milvus.env
-      required: false
     - path: ./envs/infrastructure/nginx.env
       required: false
     - path: ./envs/infrastructure/certbot.env
       required: false
     - path: ./envs/infrastructure/ssrf-proxy.env
       required: false
-    - path: ./envs/infrastructure/etcd.env
-      required: false
-    - path: ./envs/infrastructure/minio.env
-      required: false
-    - path: ./envs/infrastructure/milvus-standalone.env
-      required: false
     - ./.env
   networks:
     - ssrf_proxy_network
@@ -87,50 +49,12 @@ x-shared-worker-config: &shared-worker-config
       required: false
     - path: ./envs/vectorstores/weaviate.env
       required: false
-    - path: ./envs/vectorstores/qdrant.env
-      required: false
-    - path: ./envs/vectorstores/oceanbase.env
-      required: false
-    - path: ./envs/vectorstores/seekdb.env
-      required: false
-    - path: ./envs/vectorstores/couchbase.env
-      required: false
-    - path: ./envs/vectorstores/pgvector.env
-      required: false
-    - path: ./envs/vectorstores/vastbase.env
-      required: false
-    - path: ./envs/vectorstores/pgvecto-rs.env
-      required: false
-    - path: ./envs/vectorstores/chroma.env
-      required: false
-    - path: ./envs/vectorstores/iris.env
-      required: false
-    - path: ./envs/vectorstores/oracle.env
-      required: false
-    - path: ./envs/vectorstores/opengauss.env
-      required: false
-    - path: ./envs/vectorstores/myscale.env
-      required: false
-    - path: ./envs/vectorstores/matrixone.env
-      required: false
-    - path: ./envs/vectorstores/elasticsearch.env
-      required: false
-    - path: ./envs/vectorstores/opensearch.env
-      required: false
-    - path: ./envs/vectorstores/milvus.env
-      required: false
     - path: ./envs/infrastructure/nginx.env
       required: false
     - path: ./envs/infrastructure/certbot.env
       required: false
     - path: ./envs/infrastructure/ssrf-proxy.env
       required: false
-    - path: ./envs/infrastructure/etcd.env
-      required: false
-    - path: ./envs/infrastructure/minio.env
-      required: false
-    - path: ./envs/infrastructure/milvus-standalone.env
-      required: false
     - ./.env
   networks:
     - ssrf_proxy_network
@@ -153,50 +77,12 @@ x-shared-worker-beat-config: &shared-worker-beat-config
       required: false
     - path: ./envs/vectorstores/weaviate.env
       required: false
-    - path: ./envs/vectorstores/qdrant.env
-      required: false
-    - path: ./envs/vectorstores/oceanbase.env
-      required: false
-    - path: ./envs/vectorstores/seekdb.env
-      required: false
-    - path: ./envs/vectorstores/couchbase.env
-      required: false
-    - path: ./envs/vectorstores/pgvector.env
-      required: false
-    - path: ./envs/vectorstores/vastbase.env
-      required: false
-    - path: ./envs/vectorstores/pgvecto-rs.env
-      required: false
-    - path: ./envs/vectorstores/chroma.env
-      required: false
-    - path: ./envs/vectorstores/iris.env
-      required: false
-    - path: ./envs/vectorstores/oracle.env
-      required: false
-    - path: ./envs/vectorstores/opengauss.env
-      required: false
-    - path: ./envs/vectorstores/myscale.env
-      required: false
-    - path: ./envs/vectorstores/matrixone.env
-      required: false
-    - path: ./envs/vectorstores/elasticsearch.env
-      required: false
-    - path: ./envs/vectorstores/opensearch.env
-      required: false
-    - path: ./envs/vectorstores/milvus.env
-      required: false
     - path: ./envs/infrastructure/nginx.env
       required: false
     - path: ./envs/infrastructure/certbot.env
       required: false
     - path: ./envs/infrastructure/ssrf-proxy.env
       required: false
-    - path: ./envs/infrastructure/etcd.env
-      required: false
-    - path: ./envs/infrastructure/minio.env
-      required: false
-    - path: ./envs/infrastructure/milvus-standalone.env
-      required: false
     - ./.env
   networks:
     - ssrf_proxy_network
@@ -251,18 +137,11 @@ services:
       db_mysql:
         condition: service_healthy
         required: false
-      oceanbase:
-        condition: service_healthy
-        required: false
-      seekdb:
-        condition: service_healthy
-        required: false
       redis:
         condition: service_started
       agent_backend:
         condition: service_started
     volumes:
-      # Mount the storage directory to the container, for storing user files.
       - ./volumes/app/storage:/app/api/storage
     healthcheck:
       test: ["CMD", "curl", "-f", "http://localhost:5001/health"]
@@ -270,9 +149,6 @@ services:
       timeout: 5s
       retries: 3
       start_period: 30s
-    networks:
-      - ssrf_proxy_network
-      - default
 
   # WebSocket service for workflow collaboration.
   api_websocket:
@@ -295,12 +171,8 @@ services:
         required: false
       redis:
         condition: service_started
-    networks:
-      - ssrf_proxy_network
-      - default
 
   # worker service
-  # The Celery worker for processing all queues (dataset, workflow, mail, etc.)
   worker:
     <<: *shared-worker-config
     image: langgenius/dify-api:1.16.1
@@ -325,18 +197,11 @@ services:
       db_mysql:
         condition: service_healthy
         required: false
-      oceanbase:
-        condition: service_healthy
-        required: false
-      seekdb:
-        condition: service_healthy
-        required: false
       redis:
         condition: service_started
       agent_backend:
         condition: service_started
     volumes:
-      # Mount the storage directory to the container, for storing user files.
       - ./volumes/app/storage:/app/api/storage
     healthcheck:
       test: ["CMD-SHELL", "celery -A celery_healthcheck.celery inspect ping"]
@@ -345,12 +210,8 @@ services:
       retries: 3
       start_period: 60s
       disable: ${COMPOSE_WORKER_HEALTHCHECK_DISABLED:-true}
-    networks:
-      - ssrf_proxy_network
-      - default
 
   # worker_beat service
-  # Celery beat for scheduling periodic tasks.
   worker_beat:
     <<: *shared-worker-beat-config
     image: langgenius/dify-api:1.16.1
@@ -365,12 +226,6 @@ services:
       db_mysql:
         condition: service_healthy
         required: false
-      oceanbase:
-        condition: service_healthy
-        required: false
-      seekdb:
-        condition: service_healthy
-        required: false
       redis:
         condition: service_started
     healthcheck:
@@ -380,9 +235,6 @@ services:
       retries: 3
       start_period: 60s
       disable: ${COMPOSE_WORKER_HEALTHCHECK_DISABLED:-true}
-    networks:
-      - ssrf_proxy_network
-      - default
 
   # Frontend web application.
   web:
@@ -497,9 +349,7 @@ services:
     environment:
       REDISCLI_AUTH: ${REDIS_PASSWORD:-difyai123456}
     volumes:
-      # Mount the redis data directory to the container.
       - ./volumes/redis/data:/data
-    # Set the redis password when startup redis server.
     command: redis-server --requirepass ${REDIS_PASSWORD:-difyai123456}
     healthcheck:
       test:
@@ -519,9 +369,6 @@ services:
         required: false
       - ./.env
     environment:
-      # The DifySandbox configurations
-      # Make sure you are changing this key for your deployment with a strong key.
-      # You can generate a strong key using `openssl rand -base64 42`.
       API_KEY: ${SANDBOX_API_KEY:-dify-sandbox}
       GIN_MODE: ${SANDBOX_GIN_MODE:-release}
       WORKER_TIMEOUT: ${SANDBOX_WORKER_TIMEOUT:-15}
@@ -539,14 +386,6 @@ services:
       - ssrf_proxy_network
 
   # Local sandbox for Dify Agent shell workspaces.
-  # Network isolation: local_sandbox has NO direct route to `api`. Its only
-  # networks are `agent_sandbox_network` (so agent_backend can reach it on 5004
-  # for shellctl, and it can reach agent_backend directly) and
-  # `local_sandbox_proxy_network` (so its egress is forced through
-  # agent_ssrf_proxy).
-  # All non-agent_backend/localhost traffic goes through the Squid forward proxy
-  # on port 3128, which only allows agent_backend /agent-stub/ and the Dify API
-  # /files/* endpoints (see ssrf_proxy/squid-agent.conf.template).
   local_sandbox:
     image: langgenius/dify-agent-local-sandbox:1.16.1
     restart: always
@@ -648,12 +487,6 @@ services:
       db_mysql:
         condition: service_healthy
         required: false
-      oceanbase:
-        condition: service_healthy
-        required: false
-      seekdb:
-        condition: service_healthy
-        required: false
 
   # Dify Agent backend service.
   agent_backend:
@@ -684,9 +517,6 @@ services:
       DIFY_AGENT_E2B_SHELLCTL_PORT: ${DIFY_AGENT_E2B_SHELLCTL_PORT:-5004}
       DIFY_AGENT_SANDBOX_FILE_UPLOAD_MAX_BYTES: ${PLUGIN_MAX_FILE_SIZE:-52428800}
       DIFY_AGENT_STUB_API_BASE_URL: ${DIFY_AGENT_STUB_API_BASE_URL:-http://agent_backend:5050/agent-stub}
-      # This is security-sensitive: it derives the JWE encryption key for Agent Stub bearer tokens.
-      # Replace this development default in production.
-      # Generate one with: python -c 'import secrets; print(secrets.token_urlsafe(32))'
       DIFY_AGENT_SERVER_SECRET_KEY: ${DIFY_AGENT_SERVER_SECRET_KEY:-MDEyMzQ1Njc4OWFiY2RlZjAxMjM0NTY3ODlhYmNkZWY}
       DIFY_AGENT_API_TOKEN: ${DIFY_AGENT_API_TOKEN:-dify-agent-run-token-for-dev-only}
       DIFY_AGENT_SHUTDOWN_GRACE_SECONDS: ${DIFY_AGENT_SHUTDOWN_GRACE_SECONDS:-30}
@@ -698,8 +528,6 @@ services:
         condition: service_started
     networks:
       - default
-      # Shared internal network with local_sandbox so agent_backend can reach it
-      # on port 5004 (shellctl entrypoint) while local_sandbox stays off `default`.
       - agent_sandbox_network
 
   # Dedicated SSRF proxy for the dify-agent local_sandbox.
@@ -720,15 +548,10 @@ services:
       HTTP_PORT: ${SSRF_HTTP_PORT:-3128}
       COREDUMP_DIR: ${SSRF_COREDUMP_DIR:-/var/spool/squid}
     networks:
-      # Needs to reach api and agent_backend as forward-proxy destinations.
       - default
-      # Only agent_ssrf_proxy and local_sandbox share this internal network, so
-      # the local_sandbox can reach Squid without gaining a direct route to `api`.
       - local_sandbox_proxy_network
 
   # ssrf_proxy server
-  # for more information, please refer to
-  # https://docs.dify.ai/learn-more/faq/install-faq#18-why-is-ssrf-proxy-needed%3F
   ssrf_proxy:
     image: ubuntu/squid:latest
     restart: always
@@ -743,7 +566,6 @@ services:
         "cp /docker-entrypoint-mount.sh /docker-entrypoint.sh && sed -i 's/\r$$//' /docker-entrypoint.sh && chmod +x /docker-entrypoint.sh && /docker-entrypoint.sh",
       ]
     environment:
-      # pls clearly modify the squid env vars to fit your network environment.
       HTTP_PORT: ${SSRF_HTTP_PORT:-3128}
       COREDUMP_DIR: ${SSRF_COREDUMP_DIR:-/var/spool/squid}
       SSRF_PROXY_ALLOW_PRIVATE_IPS: ${SSRF_PROXY_ALLOW_PRIVATE_IPS:-}
@@ -753,7 +575,6 @@ services:
       - default
 
   # Certbot service
-  # use `docker-compose --profile certbot up` to start the certbot service.
   certbot:
     image: certbot/certbot
     profiles:
@@ -773,7 +594,6 @@ services:
     command: ["tail", "-f", "/dev/null"]
 
   # The nginx reverse proxy.
-  # used for reverse proxying the API service and Web service.
   nginx:
     image: nginx:latest
     restart: always
@@ -783,8 +603,8 @@ services:
       - ./nginx/https.conf.template:/etc/nginx/https.conf.template
       - ./nginx/conf.d:/etc/nginx/conf.d
       - ./nginx/docker-entrypoint.sh:/docker-entrypoint-mount.sh
-      - ./nginx/ssl:/etc/ssl # cert dir (legacy)
-      - ./volumes/certbot/conf/live:/etc/letsencrypt/live # cert dir (with certbot container)
+      - ./nginx/ssl:/etc/ssl
+      - ./volumes/certbot/conf/live:/etc/letsencrypt/live
       - ./volumes/certbot/conf:/etc/letsencrypt
       - ./volumes/certbot/www:/var/www/html
     entrypoint:
@@ -798,8 +618,6 @@ services:
       NGINX_HTTPS_ENABLED: ${NGINX_HTTPS_ENABLED:-false}
       NGINX_SSL_PORT: ${NGINX_SSL_PORT:-443}
       NGINX_PORT: ${NGINX_PORT:-80}
-      # You're required to add your own SSL certificates/keys to the `./nginx/ssl` directory
-      # and modify the env vars below in .env if HTTPS_ENABLED is true.
       NGINX_SSL_CERT_FILENAME: ${NGINX_SSL_CERT_FILENAME:-dify.crt}
       NGINX_SSL_CERT_KEY_FILENAME: ${NGINX_SSL_CERT_KEY_FILENAME:-dify.key}
       NGINX_SSL_PROTOCOLS: ${NGINX_SSL_PROTOCOLS:-TLSv1.2 TLSv1.3}
@@ -814,9 +632,9 @@ services:
     depends_on:
       - api
       - web
-    ports:
-      - "${EXPOSE_NGINX_PORT:-80}:${NGINX_PORT:-80}"
-      - "${EXPOSE_NGINX_SSL_PORT:-443}:${NGINX_SSL_PORT:-443}"
+    #ports:
+    #  - "${EXPOSE_NGINX_PORT:-80}:${NGINX_PORT:-80}"
+    #  - "${EXPOSE_NGINX_SSL_PORT:-443}:${NGINX_SSL_PORT:-443}"
 
   # The Weaviate vector store.
   weaviate:
@@ -825,11 +643,8 @@ services:
       - weaviate
     restart: always
     volumes:
-      # Mount the Weaviate data directory to the con tainer.
       - ./volumes/weaviate:/var/lib/weaviate
     environment:
-      # The Weaviate configurations
-      # You can refer to the [Weaviate](https://weaviate.io/developers/weaviate/config-refs/env-vars) documentation for more information.
       PERSISTENCE_DATA_PATH: ${WEAVIATE_PERSISTENCE_DATA_PATH:-/var/lib/weaviate}
       QUERY_DEFAULTS_LIMIT: ${WEAVIATE_QUERY_DEFAULTS_LIMIT:-25}
       AUTHENTICATION_ANONYMOUS_ACCESS_ENABLED: ${WEAVIATE_AUTHENTICATION_ANONYMOUS_ACCESS_ENABLED:-false}
@@ -845,487 +660,13 @@ services:
       ENABLE_TOKENIZER_KAGOME_JA: ${WEAVIATE_ENABLE_TOKENIZER_KAGOME_JA:-false}
       ENABLE_TOKENIZER_KAGOME_KR: ${WEAVIATE_ENABLE_TOKENIZER_KAGOME_KR:-false}
 
-  # OceanBase vector database
-  oceanbase:
-    image: oceanbase/oceanbase-ce:4.3.5-lts
-    container_name: oceanbase
-    profiles:
-      - oceanbase
-    restart: always
-    volumes:
-      - ./volumes/oceanbase/data:/root/ob
-      - ./volumes/oceanbase/conf:/root/.obd/cluster
-      - ./volumes/oceanbase/init.d:/root/boot/init.d
-    environment:
-      OB_MEMORY_LIMIT: ${OCEANBASE_MEMORY_LIMIT:-6G}
-      OB_SYS_PASSWORD: ${OCEANBASE_VECTOR_PASSWORD:-difyai123456}
-      OB_TENANT_PASSWORD: ${OCEANBASE_VECTOR_PASSWORD:-difyai123456}
-      OB_CLUSTER_NAME: ${OCEANBASE_CLUSTER_NAME:-difyai}
-      OB_SERVER_IP: 127.0.0.1
-      MODE: mini
-      LANG: C.UTF-8
-      LC_ALL: C.UTF-8
-    ports:
-      - "${OCEANBASE_VECTOR_PORT:-2881}:2881"
-    healthcheck:
-      test:
-        [
-          "CMD-SHELL",
-          'obclient -h127.0.0.1 -P2881 -uroot@test -p${OCEANBASE_VECTOR_PASSWORD:-difyai123456} -e "SELECT 1;"',
-        ]
-      interval: 10s
-      retries: 30
-      start_period: 30s
-      timeout: 10s
-
-  # seekdb vector database
-  seekdb:
-    image: oceanbase/seekdb:latest
-    container_name: seekdb
-    profiles:
-      - seekdb
-    restart: always
-    volumes:
-      - ./volumes/seekdb:/var/lib/oceanbase
-    environment:
-      ROOT_PASSWORD: ${OCEANBASE_VECTOR_PASSWORD:-difyai123456}
-      MEMORY_LIMIT: ${SEEKDB_MEMORY_LIMIT:-2G}
-      REPORTER: dify-ai-seekdb
-    ports:
-      - "${OCEANBASE_VECTOR_PORT:-2881}:2881"
-    healthcheck:
-      test:
-        [
-          "CMD-SHELL",
-          'mysql -h127.0.0.1 -P2881 -uroot -p${OCEANBASE_VECTOR_PASSWORD:-difyai123456} -e "SELECT 1;"',
-        ]
-      interval: 5s
-      retries: 60
-      timeout: 5s
-
-  # Qdrant vector store.
-  # (if used, you need to set VECTOR_STORE to qdrant in the api & worker service.)
-  qdrant:
-    image: langgenius/qdrant:v1.8.3
-    profiles:
-      - qdrant
-    restart: always
-    volumes:
-      - ./volumes/qdrant:/qdrant/storage
-    environment:
-      QDRANT_API_KEY: ${QDRANT_API_KEY:-difyai123456}
-
-  # The Couchbase vector store.
-  couchbase-server:
-    build: ./couchbase-server
-    profiles:
-      - couchbase
-    restart: always
-    environment:
-      - CLUSTER_NAME=dify_search
-      - COUCHBASE_ADMINISTRATOR_USERNAME=${COUCHBASE_USER:-Administrator}
-      - COUCHBASE_ADMINISTRATOR_PASSWORD=${COUCHBASE_PASSWORD:-password}
-      - COUCHBASE_BUCKET=${COUCHBASE_BUCKET_NAME:-Embeddings}
-      - COUCHBASE_BUCKET_RAMSIZE=512
-      - COUCHBASE_RAM_SIZE=2048
-      - COUCHBASE_EVENTING_RAM_SIZE=512
-      - COUCHBASE_INDEX_RAM_SIZE=512
-      - COUCHBASE_FTS_RAM_SIZE=1024
-    hostname: couchbase-server
-    container_name: couchbase-server
-    working_dir: /opt/couchbase
-    stdin_open: true
-    tty: true
-    entrypoint: [""]
-    command: sh -c "/opt/couchbase/init/init-cbserver.sh"
-    volumes:
-      - ./volumes/couchbase/data:/opt/couchbase/var/lib/couchbase/data
-    healthcheck:
-      # ensure bucket was created before proceeding
-      test:
-        [
-          "CMD-SHELL",
-          "curl -s -f -u Administrator:password http://localhost:8091/pools/default/buckets | grep -q '\\[{' || exit 1",
-        ]
-      interval: 10s
-      retries: 10
-      start_period: 30s
-      timeout: 10s
-
-  # The pgvector vector database.
-  pgvector:
-    image: pgvector/pgvector:pg16
-    profiles:
-      - pgvector
-    restart: always
-    environment:
-      PGUSER: ${PGVECTOR_PGUSER:-postgres}
-      # The password for the default postgres user.
-      POSTGRES_PASSWORD: ${PGVECTOR_POSTGRES_PASSWORD:-difyai123456}
-      # The name of the default postgres database.
-      POSTGRES_DB: ${PGVECTOR_POSTGRES_DB:-dify}
-      # postgres data directory
-      PGDATA: ${PGVECTOR_PGDATA:-/var/lib/postgresql/data/pgdata}
-      # pg_bigm module for full text search
-      PG_BIGM: ${PGVECTOR_PG_BIGM:-false}
-      PG_BIGM_VERSION: ${PGVECTOR_PG_BIGM_VERSION:-1.2-20240606}
-    volumes:
-      - ./volumes/pgvector/data:/var/lib/postgresql/data
-      - ./pgvector/docker-entrypoint.sh:/docker-entrypoint.sh
-    entrypoint: ["/docker-entrypoint.sh"]
-    healthcheck:
-      test: ["CMD", "pg_isready"]
-      interval: 1s
-      timeout: 3s
-      retries: 30
-
-  # get image from https://www.vastdata.com.cn/
-  vastbase:
-    image: vastdata/vastbase-vector
-    profiles:
-      - vastbase
-    restart: always
-    environment:
-      - VB_DBCOMPATIBILITY=PG
-      - VB_DB=dify
-      - VB_USERNAME=dify
-      - VB_PASSWORD=Difyai123456
-    ports:
-      - "5434:5432"
-    volumes:
-      - ./vastbase/lic:/home/vastbase/vastbase/lic
-      - ./vastbase/data:/home/vastbase/data
-      - ./vastbase/backup:/home/vastbase/backup
-      - ./vastbase/backup_log:/home/vastbase/backup_log
-    healthcheck:
-      test: ["CMD", "pg_isready"]
-      interval: 1s
-      timeout: 3s
-      retries: 30
-
-  # pgvecto-rs vector store
-  pgvecto-rs:
-    image: tensorchord/pgvecto-rs:pg16-v0.3.0
-    profiles:
-      - pgvecto-rs
-    restart: always
-    environment:
-      PGUSER: ${PGVECTOR_PGUSER:-postgres}
-      # The password for the default postgres user.
-      POSTGRES_PASSWORD: ${PGVECTOR_POSTGRES_PASSWORD:-difyai123456}
-      # The name of the default postgres database.
-      POSTGRES_DB: ${PGVECTOR_POSTGRES_DB:-dify}
-      # postgres data directory
-      PGDATA: ${PGVECTOR_PGDATA:-/var/lib/postgresql/data/pgdata}
-    volumes:
-      - ./volumes/pgvecto_rs/data:/var/lib/postgresql/data
-    healthcheck:
-      test: ["CMD", "pg_isready"]
-      interval: 1s
-      timeout: 3s
-      retries: 30
-
-  # Chroma vector database
-  chroma:
-    image: ghcr.io/chroma-core/chroma:0.5.20
-    profiles:
-      - chroma
-    restart: always
-    volumes:
-      - ./volumes/chroma:/chroma/chroma
-    environment:
-      CHROMA_SERVER_AUTHN_CREDENTIALS: ${CHROMA_SERVER_AUTHN_CREDENTIALS:-difyai123456}
-      CHROMA_SERVER_AUTHN_PROVIDER: ${CHROMA_SERVER_AUTHN_PROVIDER:-chromadb.auth.token_authn.TokenAuthenticationServerProvider}
-      IS_PERSISTENT: ${CHROMA_IS_PERSISTENT:-TRUE}
-
-  # InterSystems IRIS vector database
-  iris:
-    image: containers.intersystems.com/intersystems/iris-community:2025.3
-    profiles:
-      - iris
-    container_name: iris
-    restart: always
-    init: true
-    ports:
-      - "${IRIS_SUPER_SERVER_PORT:-1972}:1972"
-      - "${IRIS_WEB_SERVER_PORT:-52773}:52773"
-    volumes:
-      - ./volumes/iris:/durable
-      - ./iris/iris-init.script:/iris-init.script
-      - ./iris/docker-entrypoint.sh:/custom-entrypoint.sh
-    entrypoint: ["/custom-entrypoint.sh"]
-    tty: true
-    environment:
-      TZ: ${IRIS_TIMEZONE:-UTC}
-      ISC_DATA_DIRECTORY: /durable/iris
-
-  # Oracle vector database
-  oracle:
-    image: container-registry.oracle.com/database/free:latest
-    profiles:
-      - oracle
-    restart: always
-    volumes:
-      - source: oradata
-        type: volume
-        target: /opt/oracle/oradata
-      - ./startupscripts:/opt/oracle/scripts/startup
-    environment:
-      ORACLE_PWD: ${ORACLE_PWD:-Dify123456}
-      ORACLE_CHARACTERSET: ${ORACLE_CHARACTERSET:-AL32UTF8}
-
-  # Milvus vector database services
-  etcd:
-    container_name: milvus-etcd
-    image: quay.io/coreos/etcd:v3.5.5
-    profiles:
-      - milvus
-    environment:
-      ETCD_AUTO_COMPACTION_MODE: ${ETCD_AUTO_COMPACTION_MODE:-revision}
-      ETCD_AUTO_COMPACTION_RETENTION: ${ETCD_AUTO_COMPACTION_RETENTION:-1000}
-      ETCD_QUOTA_BACKEND_BYTES: ${ETCD_QUOTA_BACKEND_BYTES:-4294967296}
-      ETCD_SNAPSHOT_COUNT: ${ETCD_SNAPSHOT_COUNT:-50000}
-    volumes:
-      - ./volumes/milvus/etcd:/etcd
-    command: etcd -advertise-client-urls=http://127.0.0.1:2379 -listen-client-urls http://0.0.0.0:2379 --data-dir /etcd
-    healthcheck:
-      test: ["CMD", "etcdctl", "endpoint", "health"]
-      interval: 30s
-      timeout: 20s
-      retries: 3
-    networks:
-      - milvus
-
-  minio:
-    container_name: milvus-minio
-    image: minio/minio:RELEASE.2023-03-20T20-16-18Z
-    profiles:
-      - milvus
-    environment:
-      MINIO_ACCESS_KEY: ${MINIO_ACCESS_KEY:-minioadmin}
-      MINIO_SECRET_KEY: ${MINIO_SECRET_KEY:-minioadmin}
-    volumes:
-      - ./volumes/milvus/minio:/minio_data
-    command: minio server /minio_data --console-address ":9001"
-    healthcheck:
-      test: ["CMD", "curl", "-f", "http://localhost:9000/minio/health/live"]
-      interval: 30s
-      timeout: 20s
-      retries: 3
-    networks:
-      - milvus
-
-  milvus-standalone:
-    container_name: milvus-standalone
-    image: milvusdb/milvus:v2.6.3
-    profiles:
-      - milvus
-    command: ["milvus", "run", "standalone"]
-    environment:
-      ETCD_ENDPOINTS: ${ETCD_ENDPOINTS:-etcd:2379}
-      MINIO_ADDRESS: ${MINIO_ADDRESS:-minio:9000}
-      common.security.authorizationEnabled: ${MILVUS_AUTHORIZATION_ENABLED:-true}
-    volumes:
-      - ./volumes/milvus/milvus:/var/lib/milvus
-    healthcheck:
-      test: ["CMD", "curl", "-f", "http://localhost:9091/healthz"]
-      interval: 30s
-      start_period: 90s
-      timeout: 20s
-      retries: 3
-    depends_on:
-      - etcd
-      - minio
-    ports:
-      - 19530:19530
-      - 9091:9091
-    networks:
-      - milvus
-
-  # Opensearch vector database
-  opensearch:
-    container_name: opensearch
-    image: opensearchproject/opensearch:latest
-    profiles:
-      - opensearch
-    environment:
-      discovery.type: ${OPENSEARCH_DISCOVERY_TYPE:-single-node}
-      bootstrap.memory_lock: ${OPENSEARCH_BOOTSTRAP_MEMORY_LOCK:-true}
-      OPENSEARCH_JAVA_OPTS: -Xms${OPENSEARCH_JAVA_OPTS_MIN:-512m} -Xmx${OPENSEARCH_JAVA_OPTS_MAX:-1024m}
-      OPENSEARCH_INITIAL_ADMIN_PASSWORD: ${OPENSEARCH_INITIAL_ADMIN_PASSWORD:-Qazwsxedc!@#123}
-    ulimits:
-      memlock:
-        soft: ${OPENSEARCH_MEMLOCK_SOFT:--1}
-        hard: ${OPENSEARCH_MEMLOCK_HARD:--1}
-      nofile:
-        soft: ${OPENSEARCH_NOFILE_SOFT:-65536}
-        hard: ${OPENSEARCH_NOFILE_HARD:-65536}
-    volumes:
-      - ./volumes/opensearch/data:/usr/share/opensearch/data
-    networks:
-      - opensearch-net
-
-  opensearch-dashboards:
-    container_name: opensearch-dashboards
-    image: opensearchproject/opensearch-dashboards:latest
-    profiles:
-      - opensearch
-    environment:
-      OPENSEARCH_HOSTS: '["https://opensearch:9200"]'
-    volumes:
-      - ./volumes/opensearch/opensearch_dashboards.yml:/usr/share/opensearch-dashboards/config/opensearch_dashboards.yml
-    networks:
-      - opensearch-net
-    depends_on:
-      - opensearch
-
-  # opengauss vector database.
-  opengauss:
-    image: opengauss/opengauss:7.0.0-RC1
-    profiles:
-      - opengauss
-    privileged: true
-    restart: always
-    environment:
-      GS_USERNAME: ${OPENGAUSS_USER:-postgres}
-      GS_PASSWORD: ${OPENGAUSS_PASSWORD:-Dify@123}
-      GS_PORT: ${OPENGAUSS_PORT:-6600}
-      GS_DB: ${OPENGAUSS_DATABASE:-dify}
-    volumes:
-      - ./volumes/opengauss/data:/var/lib/opengauss/data
-    healthcheck:
-      test: ["CMD-SHELL", "netstat -lntp | grep tcp6 > /dev/null 2>&1"]
-      interval: 10s
-      timeout: 10s
-      retries: 10
-    ports:
-      - ${OPENGAUSS_PORT:-6600}:${OPENGAUSS_PORT:-6600}
-
-  # MyScale vector database
-  myscale:
-    container_name: myscale
-    image: myscale/myscaledb:1.6.4
-    profiles:
-      - myscale
-    restart: always
-    tty: true
-    volumes:
-      - ./volumes/myscale/data:/var/lib/clickhouse
-      - ./volumes/myscale/log:/var/log/clickhouse-server
-      - ./volumes/myscale/config/users.d/custom_users_config.xml:/etc/clickhouse-server/users.d/custom_users_config.xml
-    ports:
-      - ${MYSCALE_PORT:-8123}:${MYSCALE_PORT:-8123}
-
-  # Matrixone vector store.
-  matrixone:
-    hostname: matrixone
-    image: matrixorigin/matrixone:2.1.1
-    profiles:
-      - matrixone
-    restart: always
-    volumes:
-      - ./volumes/matrixone/data:/mo-data
-    ports:
-      - ${MATRIXONE_PORT:-6001}:${MATRIXONE_PORT:-6001}
-
-  # https://www.elastic.co/guide/en/elasticsearch/reference/current/settings.html
-  # https://www.elastic.co/guide/en/elasticsearch/reference/current/docker.html#docker-prod-prerequisites
-  elasticsearch:
-    image: docker.elastic.co/elasticsearch/elasticsearch:8.14.3
-    container_name: elasticsearch
-    profiles:
-      - elasticsearch
-      - elasticsearch-ja
-    restart: always
-    volumes:
-      - ./elasticsearch/docker-entrypoint.sh:/docker-entrypoint-mount.sh
-      - dify_es01_data:/usr/share/elasticsearch/data
-    environment:
-      ELASTIC_PASSWORD: ${ELASTICSEARCH_PASSWORD:-elastic}
-      VECTOR_STORE: ${VECTOR_STORE:-}
-      cluster.name: dify-es-cluster
-      node.name: dify-es0
-      discovery.type: single-node
-      xpack.license.self_generated.type: basic
-      xpack.security.enabled: "true"
-      xpack.security.enrollment.enabled: "false"
-      xpack.security.http.ssl.enabled: "false"
-    ports:
-      - ${ELASTICSEARCH_PORT:-9200}:9200
-    deploy:
-      resources:
-        limits:
-          memory: 2g
-    entrypoint: ["sh", "-c", "sh /docker-entrypoint-mount.sh"]
-    healthcheck:
-      test:
-        ["CMD", "curl", "-s", "http://localhost:9200/_cluster/health?pretty"]
-      interval: 30s
-      timeout: 10s
-      retries: 50
-
-  # https://www.elastic.co/guide/en/kibana/current/docker.html
-  # https://www.elastic.co/guide/en/kibana/current/settings.html
-  kibana:
-    image: docker.elastic.co/kibana/kibana:8.14.3
-    container_name: kibana
-    profiles:
-      - elasticsearch
-    depends_on:
-      - elasticsearch
-    restart: always
-    environment:
-      XPACK_ENCRYPTEDSAVEDOBJECTS_ENCRYPTIONKEY: d1a66dfd-c4d3-4a0a-8290-2abcb83ab3aa
-      NO_PROXY: localhost,127.0.0.1,elasticsearch,kibana
-      XPACK_SECURITY_ENABLED: "true"
-      XPACK_SECURITY_ENROLLMENT_ENABLED: "false"
-      XPACK_SECURITY_HTTP_SSL_ENABLED: "false"
-      XPACK_FLEET_ISAIRGAPPED: "true"
-      I18N_LOCALE: zh-CN
-      SERVER_PORT: "5601"
-      ELASTICSEARCH_HOSTS: http://elasticsearch:9200
-    ports:
-      - ${KIBANA_PORT:-5601}:5601
-    healthcheck:
-      test: ["CMD-SHELL", "curl -s http://localhost:5601 >/dev/null || exit 1"]
-      interval: 30s
-      timeout: 10s
-      retries: 3
-
-  # unstructured .
-  # (if used, you need to set ETL_TYPE to Unstructured in the api & worker service.)
-  unstructured:
-    image: downloads.unstructured.io/unstructured-io/unstructured-api:latest
-    profiles:
-      - unstructured
-    restart: always
-    volumes:
-      - ./volumes/unstructured:/app/data
-
 networks:
-  # create a network between sandbox, api and ssrf_proxy, and can not access outside.
   ssrf_proxy_network:
     driver: bridge
     internal: true
-  # Internal network shared only by agent_ssrf_proxy and local_sandbox.
   local_sandbox_proxy_network:
     driver: bridge
     internal: true
-  # shellctl control channel (agent_backend -> local_sandbox:5004).
-  # sandbox can access agent backend through this network, this is
-  # a known limitation.
-  #
-  # The agent runtime respects HTTP(S)_PROXY, but arbitrary code execution
-  # is still possible through the shellctl channel.
   agent_sandbox_network:
     driver: bridge
-    internal: true
-  milvus:
-    driver: bridge
-  opensearch-net:
-    driver: bridge
-    internal: true
-
-volumes:
-  oradata:
-  dify_es01_data:
+    internal: true

e2e/.env.example

@@ -22,6 +22,8 @@ E2E_MODEL_PROVIDER_CREDENTIALS_JSON='{"openai_api_key":"replace-with-real-key"}'
 # from @external-model because other external-model scenarios may not use dify-agent.
 # When enabled, the E2E runner also starts the shellctl local sandbox required by
 # dify-agent's dify.shell/config runtime layer.
+# Use E2E_AGENT_BACKEND_URL only for an already-running backend; do not set it
+# together with E2E_START_AGENT_BACKEND.
 # E2E_START_AGENT_BACKEND=1
 # E2E_AGENT_BACKEND_PORT=5050
 # E2E_AGENT_BACKEND_URL=http://127.0.0.1:5050

e2e/AGENTS.md

@@ -8,9 +8,11 @@ Run commands from the repository root. Install dependencies and browsers once wi
 
 - Existing initialized instance: `pnpm -C e2e e2e`
 - Reset, initialize, and run deterministic scenarios: `pnpm -C e2e e2e:full`
+- Prepare and run scenarios backed by shared fixtures: `E2E_START_AGENT_BACKEND=1 pnpm -C e2e e2e:prepared`
 - Tagged subset: `pnpm -C e2e e2e -- --tags @smoke`
 - Headed debugging: `pnpm -C e2e e2e:headed -- --tags @smoke`
-- External runtime preparation and run: `pnpm -C e2e e2e:external:prepare`, then `pnpm -C e2e e2e:external`
+- Prepare and run external runtime scenarios: `E2E_START_AGENT_BACKEND=1 pnpm -C e2e e2e:external`
+- Seed against existing middleware without running Cucumber: `pnpm -C e2e seed -- --profile <prepared|external-runtime|post-merge>`
 - Reset persisted E2E state: `pnpm -C e2e e2e:reset`
 - Middleware lifecycle: `pnpm -C e2e e2e:middleware:up` and `pnpm -C e2e e2e:middleware:down`
 - Scoped static checks: `vp check e2e`
@@ -20,7 +22,8 @@ The runner reuses `web/.next/BUILD_ID` when present. Set `E2E_FORCE_WEB_BUILD=1`
 ## Runtime Ownership
 
 - `scripts/setup.ts` owns reset, middleware, backend, and frontend startup.
-- `scripts/run-cucumber.ts` owns E2E orchestration and Cucumber invocation.
+- `scripts/run-cucumber.ts` is the only E2E runtime orchestrator. It owns service lifetime, optional seed execution, Cucumber invocation, and teardown.
+- `scripts/seed-runner.ts` owns fixture creation and verification against an already-running runtime; it never starts services.
 - `support/web-server.ts` owns frontend reuse, readiness, and shutdown.
 - `features/support/hooks.ts` owns shared auth bootstrap, scenario lifecycle, and diagnostics.
 - `features/support/world.ts` owns `DifyWorld`, the per-scenario behavior `BrowserContext`, and its authenticated setup and cleanup client. Browser and API identities remain separate so unauthenticated and logout journeys cannot invalidate fixture ownership.
@@ -33,12 +36,14 @@ An uninitialized instance is installed and authenticated lazily; an initialized
 ## Tags And External Runtime
 
 - Default scenarios use shared authenticated storage state. `@unauthenticated` creates a clean context; `@authenticated` is an intent and selection tag only.
-- `@prepared` requires the strict post-merge seed profile.
+- `@prepared` requires the prepared fixtures; the post-merge seed profile includes them.
 - `@external-model` and `@external-tool` identify scenarios that call real external runtimes. Deterministic commands exclude these tags; external commands are opt-in.
 - `@microphone` uses the checked-in fake audio fixture and an isolated Chromium context.
 - `@browser-smoke` runs focused keyboard and navigation coverage in Chromium and WebKit CI lanes.
 - Feature-owned services use their own tags. Agent v2 runtime scenarios use `@agent-backend-runtime` and require the explicit runtime-availability step. Set `E2E_START_AGENT_BACKEND=1` to start it locally, or provide `E2E_AGENT_BACKEND_URL` / `AGENT_BACKEND_BASE_URL`.
 
+Seed and Cucumber must share one runtime lifecycle. Combined commands own reset, middleware, services, seed, Cucumber, and teardown; CI must not reproduce that lifecycle in workflow YAML. `E2E_START_AGENT_BACKEND=1` starts a managed local backend before the API; it is mutually exclusive with an explicit Agent backend URL.
+
 Do not overload runtime tags to imply unrelated services or silently skip behavior when a required fixture is missing.
 
 ## Browser, API, And Contract Boundaries

e2e/features/agent-v2/AGENTS.md

@@ -48,16 +48,15 @@ Use `the Agent v2 configuration should be saved automatically` for Configure aut
 
 ## Seed and fixture contract
 
-Seed scripts create or update environment-owned models, plugins, datasets, Agents, and workflows. `fixtures.steps.ts` resolves and validates those resources before a dependent behavior runs. Missing, inactive, unindexed, or drifted fixtures must throw and fail the scenario; never return `skipped`.
+Seed tasks create or update environment-owned models, plugins, datasets, Agents, and workflows. `fixtures.steps.ts` resolves and validates those resources before a dependent behavior runs. Missing, inactive, unindexed, or drifted fixtures must throw and fail the scenario; never return `skipped`.
 
 `@prepared` scenarios are excluded from deterministic PR core. Post-merge runs:
 
 ```bash
-pnpm -C e2e e2e:post-merge:prepare
-pnpm -C e2e e2e:post-merge
+E2E_START_AGENT_BACKEND=1 pnpm -C e2e e2e:post-merge
 ```
 
-The strict seed must finish without blocked tasks. The concrete resource inventory and defaults belong to the seed profile and environment configuration rather than this guidance.
+The command owns runtime setup, strict seed, Cucumber, and teardown. The strict seed must finish without blocked tasks. The concrete resource inventory and defaults belong to the seed profile and environment configuration rather than this guidance.
 
 Organize fixture helpers by the product resource or infrastructure capability they own, not by the feature file that happens to consume them. Keep runtime readiness adapters separate from Console resource fixtures, and keep all fixture state in the current `SeedContext` or scenario `DifyWorld` rather than module globals.
 

e2e/package.json

@@ -5,7 +5,7 @@
   "scripts": {
     "e2e": "tsx ./scripts/run-cucumber.ts",
     "e2e:external": "tsx ./scripts/run-external-runtime.ts",
-    "e2e:external:prepare": "tsx ./scripts/prepare-external-runtime.ts",
+    "e2e:external:prepare": "tsx ./scripts/run-cucumber.ts --seed-only --profile external-runtime",
     "e2e:full": "tsx ./scripts/run-cucumber.ts --full",
     "e2e:full:headed": "tsx ./scripts/run-cucumber.ts --full --headed",
     "e2e:headed": "tsx ./scripts/run-cucumber.ts --headed",
@@ -13,9 +13,11 @@
     "e2e:middleware:down": "tsx ./scripts/setup.ts middleware-down",
     "e2e:middleware:up": "tsx ./scripts/setup.ts middleware-up",
     "e2e:post-merge": "tsx ./scripts/run-post-merge.ts",
-    "e2e:post-merge:prepare": "tsx ./scripts/seed.ts --pack agent-v2 --profile post-merge",
+    "e2e:post-merge:prepare": "tsx ./scripts/run-cucumber.ts --seed-only --profile post-merge",
+    "e2e:prepared": "tsx ./scripts/run-prepared.ts",
+    "e2e:prepared:prepare": "tsx ./scripts/run-cucumber.ts --seed-only --profile prepared",
     "e2e:reset": "tsx ./scripts/setup.ts reset",
-    "seed": "tsx ./scripts/seed.ts",
+    "seed": "tsx ./scripts/run-cucumber.ts --seed-only",
     "test:unit": "vitest run",
     "type-check": "tsc"
   },

e2e/scripts/prepare-external-runtime.ts

@@ -1,17 +0,0 @@
-import { e2eDir, isMainModule, runCommandOrThrow } from './common'
-import './env-register'
-
-const main = async () => {
-  await runCommandOrThrow({
-    command: 'npx',
-    args: ['tsx', './scripts/seed.ts', '--pack', 'agent-v2', '--profile', 'external-runtime'],
-    cwd: e2eDir,
-  })
-}
-
-if (isMainModule(import.meta.url)) {
-  void main().catch((error) => {
-    console.error(error instanceof Error ? error.message : String(error))
-    process.exit(1)
-  })
-}

e2e/scripts/run-cucumber.ts

@@ -7,56 +7,16 @@ import { startLoggedProcess, stopManagedProcess, waitForUrl } from '../support/p
 import { startWebServer, stopWebServer } from '../support/web-server'
 import { apiURL, baseURL, reuseExistingWebServer } from '../test-env'
 import { e2eDir, isMainModule, runCommand } from './common'
+import { parseRunOptions, shouldStartManagedAgentBackend } from './run-options'
+import { runSeed } from './seed-runner'
 import { resetState, startMiddleware, stopMiddleware } from './setup'
 import './env-register'
 
-type RunOptions = {
-  forwardArgs: string[]
-  full: boolean
-  headed: boolean
-}
-
-const parseArgs = (argv: string[]): RunOptions => {
-  let full = false
-  let headed = false
-  const forwardArgs: string[] = []
-
-  for (const [index, arg] of argv.entries()) {
-    if (arg === '--') {
-      forwardArgs.push(...argv.slice(index + 1))
-      return { forwardArgs, full, headed }
-    }
-
-    if (arg === '--full') {
-      full = true
-      continue
-    }
-
-    if (arg === '--headed') {
-      headed = true
-      continue
-    }
-
-    forwardArgs.push(arg)
-  }
-
-  return { forwardArgs, full, headed }
-}
-
 const hasCustomTags = (forwardArgs: string[]) =>
   forwardArgs.some((arg) => arg === '--tags' || arg.startsWith('--tags='))
 
 const fullNonExternalTags = 'not @prepared and not @external-model and not @external-tool'
-
-const isTruthyEnv = (value: string | undefined) => value === '1' || value === 'true'
-
-const shouldStartAgentBackend = () => {
-  if (isTruthyEnv(process.env.E2E_START_AGENT_BACKEND)) return true
-
-  if (process.env.E2E_AGENT_BACKEND_URL || process.env.AGENT_BACKEND_BASE_URL) return false
-
-  return false
-}
+const seedCeleryQueues = 'dataset,priority_dataset,workflow_based_app_execution'
 
 const readLogTail = async (logFilePath: string) => {
   const content = await readFile(logFilePath, 'utf8').catch(() => '')
@@ -87,65 +47,41 @@ const waitForUnexpectedProcessExit = async (
   throw new Error(`${label} exited before becoming ready. See ${logFilePath}.${logTailMessage}`)
 }
 
+const waitForManagedProcess = async ({
+  errorMessage,
+  managedProcess,
+  url,
+}: {
+  errorMessage: string
+  managedProcess: ManagedProcess
+  url: string
+}) => {
+  let waiting = true
+  try {
+    await Promise.race([
+      waitForUrl(url, 180_000, 1_000),
+      waitForUnexpectedProcessExit(managedProcess, () => !waiting),
+    ])
+  } catch (error) {
+    if (error instanceof Error && error.message.includes('exited before becoming ready'))
+      throw error
+
+    throw new Error(`${errorMessage} See ${managedProcess.logFilePath}.`)
+  } finally {
+    waiting = false
+  }
+}
+
 const main = async () => {
-  const { forwardArgs, full, headed } = parseArgs(process.argv.slice(2))
-  const startMiddlewareForRun = full
-  const resetStateForRun = full
-  const startAgentBackendForRun = shouldStartAgentBackend()
-
-  if (resetStateForRun) await resetState()
-
-  if (startMiddlewareForRun) await startMiddleware()
-
+  const { forwardArgs, full, headed, seed, seedOnly } = parseRunOptions(process.argv.slice(2))
+  const startAgentBackendForRun = shouldStartManagedAgentBackend()
   const cucumberReportDir = path.join(e2eDir, 'cucumber-report')
   const logDir = path.join(e2eDir, '.logs')
-
-  await rm(cucumberReportDir, { force: true, recursive: true })
-  await mkdir(logDir, { recursive: true })
-
-  const shellctlProcess = startAgentBackendForRun
-    ? await startLoggedProcess({
-        command: 'npx',
-        args: ['tsx', './scripts/setup.ts', 'shellctl-sandbox'],
-        cwd: e2eDir,
-        label: 'shellctl sandbox',
-        logFilePath: path.join(logDir, 'cucumber-shellctl-sandbox.log'),
-      })
-    : undefined
-
-  const difyAgentProcess = startAgentBackendForRun
-    ? await startLoggedProcess({
-        command: 'npx',
-        args: ['tsx', './scripts/setup.ts', 'agent-backend'],
-        cwd: e2eDir,
-        env: {
-          E2E_START_AGENT_BACKEND: '1',
-        },
-        label: 'agent backend',
-        logFilePath: path.join(logDir, 'cucumber-agent-backend.log'),
-      })
-    : undefined
-
-  const apiProcess = await startLoggedProcess({
-    command: 'npx',
-    args: ['tsx', './scripts/setup.ts', 'api'],
-    cwd: e2eDir,
-    env: startAgentBackendForRun
-      ? {
-          E2E_START_AGENT_BACKEND: '1',
-        }
-      : undefined,
-    label: 'api server',
-    logFilePath: path.join(logDir, 'cucumber-api.log'),
-  })
-
-  const celeryProcess = await startLoggedProcess({
-    command: 'npx',
-    args: ['tsx', './scripts/setup.ts', 'celery'],
-    cwd: e2eDir,
-    label: 'celery worker',
-    logFilePath: path.join(logDir, 'cucumber-celery.log'),
-  })
+  let apiProcess: ManagedProcess | undefined
+  let celeryProcess: ManagedProcess | undefined
+  let difyAgentProcess: ManagedProcess | undefined
+  let middlewareStarted = false
+  let shellctlProcess: ManagedProcess | undefined
 
   let cleanupPromise: Promise<void> | undefined
   const cleanup = async () => {
@@ -157,7 +93,7 @@ const main = async () => {
           { label: 'Stop API server', run: () => stopManagedProcess(apiProcess) },
           { label: 'Stop agent backend', run: () => stopManagedProcess(difyAgentProcess) },
           { label: 'Stop shellctl sandbox', run: () => stopManagedProcess(shellctlProcess) },
-          ...(startMiddlewareForRun ? [{ label: 'Stop middleware', run: stopMiddleware }] : []),
+          ...(middlewareStarted ? [{ label: 'Stop middleware', run: stopMiddleware }] : []),
         ])
 
         if (cleanupErrors.length > 0)
@@ -182,60 +118,73 @@ const main = async () => {
   process.once('SIGTERM', onTerminate)
 
   try {
-    if (shellctlProcess) {
-      let waitingForShellctl = true
-      try {
-        const shellctlPort = process.env.E2E_SHELLCTL_PORT || '5004'
-        await Promise.race([
-          waitForUrl(`http://127.0.0.1:${shellctlPort}/healthz`, 180_000, 1_000),
-          waitForUnexpectedProcessExit(shellctlProcess, () => !waitingForShellctl),
-        ])
-      } catch (error) {
-        if (error instanceof Error && error.message.includes('exited before becoming ready'))
-          throw error
+    if (full) await resetState()
 
-        throw new Error(
-          `Shellctl sandbox did not become ready. See ${shellctlProcess.logFilePath}.`,
-        )
-      } finally {
-        waitingForShellctl = false
-      }
+    if (full) {
+      middlewareStarted = true
+      await startMiddleware()
     }
 
-    if (difyAgentProcess) {
-      let waitingForAgentBackend = true
-      try {
-        const agentBackendPort = process.env.E2E_AGENT_BACKEND_PORT || '5050'
-        await Promise.race([
-          waitForUrl(`http://127.0.0.1:${agentBackendPort}/openapi.json`, 180_000, 1_000),
-          waitForUnexpectedProcessExit(difyAgentProcess, () => !waitingForAgentBackend),
-        ])
-      } catch (error) {
-        if (error instanceof Error && error.message.includes('exited before becoming ready'))
-          throw error
+    if (!seedOnly) await rm(cucumberReportDir, { force: true, recursive: true })
+    await mkdir(logDir, { recursive: true })
 
-        throw new Error(`Agent backend did not become ready. See ${difyAgentProcess.logFilePath}.`)
-      } finally {
-        waitingForAgentBackend = false
-      }
+    if (startAgentBackendForRun) {
+      shellctlProcess = await startLoggedProcess({
+        command: 'npx',
+        args: ['tsx', './scripts/setup.ts', 'shellctl-sandbox'],
+        cwd: e2eDir,
+        label: 'shellctl sandbox',
+        logFilePath: path.join(logDir, 'cucumber-shellctl-sandbox.log'),
+      })
+      const shellctlPort = process.env.E2E_SHELLCTL_PORT || '5004'
+      await waitForManagedProcess({
+        errorMessage: 'Shellctl sandbox did not become ready.',
+        managedProcess: shellctlProcess,
+        url: `http://127.0.0.1:${shellctlPort}/healthz`,
+      })
+
+      difyAgentProcess = await startLoggedProcess({
+        command: 'npx',
+        args: ['tsx', './scripts/setup.ts', 'agent-backend'],
+        cwd: e2eDir,
+        env: { E2E_START_AGENT_BACKEND: '1' },
+        label: 'agent backend',
+        logFilePath: path.join(logDir, 'cucumber-agent-backend.log'),
+      })
+      const agentBackendPort = process.env.E2E_AGENT_BACKEND_PORT || '5050'
+      await waitForManagedProcess({
+        errorMessage: 'Agent backend did not become ready.',
+        managedProcess: difyAgentProcess,
+        url: `http://127.0.0.1:${agentBackendPort}/openapi.json`,
+      })
     }
 
-    let waitingForApi = true
-    try {
-      await Promise.race([
-        waitForUrl(`${apiURL}/health`, 180_000, 1_000),
-        waitForUnexpectedProcessExit(apiProcess, () => !waitingForApi),
-      ])
-    } catch (error) {
-      if (error instanceof Error && error.message.includes('exited before becoming ready'))
-        throw error
+    apiProcess = await startLoggedProcess({
+      command: 'npx',
+      args: ['tsx', './scripts/setup.ts', 'api'],
+      cwd: e2eDir,
+      env: startAgentBackendForRun ? { E2E_START_AGENT_BACKEND: '1' } : undefined,
+      label: 'api server',
+      logFilePath: path.join(logDir, 'cucumber-api.log'),
+    })
+    await waitForManagedProcess({
+      errorMessage: `API did not become ready at ${apiURL}/health.`,
+      managedProcess: apiProcess,
+      url: `${apiURL}/health`,
+    })
 
-      throw new Error(
-        `API did not become ready at ${apiURL}/health. See ${apiProcess.logFilePath}.`,
-      )
-    } finally {
-      waitingForApi = false
-    }
+    celeryProcess = await startLoggedProcess({
+      command: 'npx',
+      args: [
+        'tsx',
+        './scripts/setup.ts',
+        'celery',
+        ...(seed ? ['--queues', seedCeleryQueues] : []),
+      ],
+      cwd: e2eDir,
+      label: 'celery worker',
+      logFilePath: path.join(logDir, 'cucumber-celery.log'),
+    })
 
     await startWebServer({
       baseURL,
@@ -247,33 +196,36 @@ const main = async () => {
       timeoutMs: 300_000,
     })
 
-    const cucumberEnv: NodeJS.ProcessEnv = {
-      ...process.env,
-      CUCUMBER_HEADLESS: headed ? '0' : '1',
+    if (seed) await runSeed(seed)
+
+    if (!seedOnly) {
+      const cucumberEnv: NodeJS.ProcessEnv = {
+        ...process.env,
+        CUCUMBER_HEADLESS: headed ? '0' : '1',
+      }
+
+      if (full && !hasCustomTags(forwardArgs)) cucumberEnv.E2E_CUCUMBER_TAGS = fullNonExternalTags
+
+      const result = await runCommand({
+        command: 'npx',
+        args: [
+          'tsx',
+          './node_modules/@cucumber/cucumber/bin/cucumber.js',
+          '--config',
+          './cucumber.config.ts',
+          ...forwardArgs,
+        ],
+        cwd: e2eDir,
+        env: cucumberEnv,
+      })
+
+      if (result.exitCode === 0) {
+        const messages = await readFile(path.join(cucumberReportDir, 'report.ndjson'), 'utf8')
+        assertCucumberScenariosStarted(messages)
+      }
+
+      process.exitCode = result.exitCode
     }
-
-    if (startMiddlewareForRun && !hasCustomTags(forwardArgs))
-      cucumberEnv.E2E_CUCUMBER_TAGS = fullNonExternalTags
-
-    const result = await runCommand({
-      command: 'npx',
-      args: [
-        'tsx',
-        './node_modules/@cucumber/cucumber/bin/cucumber.js',
-        '--config',
-        './cucumber.config.ts',
-        ...forwardArgs,
-      ],
-      cwd: e2eDir,
-      env: cucumberEnv,
-    })
-
-    if (result.exitCode === 0) {
-      const messages = await readFile(path.join(cucumberReportDir, 'report.ndjson'), 'utf8')
-      assertCucumberScenariosStarted(messages)
-    }
-
-    process.exitCode = result.exitCode
   } finally {
     process.off('SIGINT', onTerminate)
     process.off('SIGTERM', onTerminate)

e2e/scripts/run-external-runtime.ts

@@ -6,7 +6,16 @@ const defaultExternalRuntimeTags = '@external-model or @external-tool'
 const main = async () => {
   await runForegroundProcess({
     command: 'npx',
-    args: ['tsx', './scripts/run-cucumber.ts', '--', '--tags', defaultExternalRuntimeTags],
+    args: [
+      'tsx',
+      './scripts/run-cucumber.ts',
+      '--full',
+      '--profile',
+      'external-runtime',
+      '--',
+      '--tags',
+      defaultExternalRuntimeTags,
+    ],
     cwd: e2eDir,
   })
 }

e2e/scripts/run-options.ts

@@ -0,0 +1,122 @@
+import type { SeedOptions } from './seed-runner'
+
+export type RunOptions = {
+  forwardArgs: string[]
+  full: boolean
+  headed: boolean
+  seed?: SeedOptions
+  seedOnly: boolean
+}
+
+const readOptionValue = (argv: string[], index: number, option: string) => {
+  const value = argv[index + 1]
+  if (!value || value.startsWith('--')) throw new Error(`${option} requires a value.`)
+
+  return value
+}
+
+export const parseRunOptions = (argv: string[]): RunOptions => {
+  let allowBlocked = false
+  let dryRun = false
+  let full = false
+  let headed = false
+  let pack = 'agent-v2'
+  let profile: string | undefined
+  let seedOnly = false
+  const forwardArgs: string[] = []
+
+  for (let index = 0; index < argv.length; index += 1) {
+    const arg = argv[index]!
+
+    if (arg === '--') {
+      forwardArgs.push(...argv.slice(index + 1))
+      break
+    }
+
+    if (arg === '--full') {
+      full = true
+      continue
+    }
+
+    if (arg === '--headed') {
+      headed = true
+      continue
+    }
+
+    if (arg === '--seed-only') {
+      seedOnly = true
+      continue
+    }
+
+    if (arg === '--allow-blocked') {
+      allowBlocked = true
+      continue
+    }
+
+    if (arg === '--dry-run') {
+      dryRun = true
+      continue
+    }
+
+    if (arg === '--pack') {
+      pack = readOptionValue(argv, index, '--pack')
+      index += 1
+      continue
+    }
+
+    if (arg.startsWith('--pack=')) {
+      pack = arg.slice('--pack='.length)
+      if (!pack) throw new Error('--pack requires a value.')
+      continue
+    }
+
+    if (arg === '--profile') {
+      profile = readOptionValue(argv, index, '--profile')
+      index += 1
+      continue
+    }
+
+    if (arg.startsWith('--profile=')) {
+      profile = arg.slice('--profile='.length)
+      if (!profile) throw new Error('--profile requires a value.')
+      continue
+    }
+
+    forwardArgs.push(arg)
+  }
+
+  const shouldSeed = seedOnly || profile !== undefined
+  if (!shouldSeed && (allowBlocked || dryRun || pack !== 'agent-v2'))
+    throw new Error('Seed options require --seed-only or --profile.')
+  if (dryRun && !seedOnly) throw new Error('--dry-run requires --seed-only.')
+
+  return {
+    forwardArgs,
+    full,
+    headed,
+    seed: shouldSeed
+      ? {
+          allowBlocked,
+          dryRun,
+          pack,
+          profile: profile ?? 'post-merge',
+        }
+      : undefined,
+    seedOnly,
+  }
+}
+
+const isTruthyEnv = (value: string | undefined) => value === '1' || value === 'true'
+
+export const shouldStartManagedAgentBackend = (env: NodeJS.ProcessEnv = process.env) => {
+  const shouldStart = isTruthyEnv(env.E2E_START_AGENT_BACKEND)
+  const externalUrl = env.E2E_AGENT_BACKEND_URL?.trim() || env.AGENT_BACKEND_BASE_URL?.trim()
+
+  if (shouldStart && externalUrl) {
+    throw new Error(
+      'E2E_START_AGENT_BACKEND cannot be enabled when E2E_AGENT_BACKEND_URL or AGENT_BACKEND_BASE_URL is set.',
+    )
+  }
+
+  return shouldStart
+}

e2e/scripts/run-post-merge.ts

@@ -6,7 +6,16 @@ const postMergeTags = '@prepared or @external-model or @external-tool'
 const main = async () => {
   await runForegroundProcess({
     command: 'npx',
-    args: ['tsx', './scripts/run-cucumber.ts', '--', '--tags', postMergeTags],
+    args: [
+      'tsx',
+      './scripts/run-cucumber.ts',
+      '--full',
+      '--profile',
+      'post-merge',
+      '--',
+      '--tags',
+      postMergeTags,
+    ],
     cwd: e2eDir,
   })
 }

e2e/scripts/run-prepared.ts

@@ -0,0 +1,23 @@
+import { e2eDir, isMainModule, runForegroundProcess } from './common'
+import './env-register'
+
+const preparedTags = '@prepared'
+
+const main = async () => {
+  await runForegroundProcess({
+    command: 'npx',
+    args: [
+      'tsx',
+      './scripts/run-cucumber.ts',
+      '--full',
+      '--profile',
+      'prepared',
+      '--',
+      '--tags',
+      preparedTags,
+    ],
+    cwd: e2eDir,
+  })
+}
+
+if (isMainModule(import.meta.url)) void main()

e2e/scripts/seed-runner.ts

@@ -0,0 +1,54 @@
+import { chromium } from '@playwright/test'
+import { createAgentV2SeedTasks } from '../features/agent-v2/support/seed'
+import { ensureAuthenticatedState } from '../fixtures/auth'
+import { createStandaloneConsoleSession } from '../support/api/console-session'
+import { runSeedTasks, writeSeedReport } from '../support/seed'
+import { baseURL } from '../test-env'
+
+export type SeedOptions = {
+  allowBlocked: boolean
+  dryRun: boolean
+  pack: string
+  profile: string
+}
+
+const getTasks = (pack: string, profile: string) => {
+  if (pack === 'agent-v2') return createAgentV2SeedTasks(profile)
+
+  throw new Error(`Unknown seed pack "${pack}".`)
+}
+
+const ensureAuth = async () => {
+  const browser = await chromium.launch({ headless: true })
+  try {
+    await ensureAuthenticatedState(browser, baseURL)
+  } finally {
+    await browser.close()
+  }
+}
+
+export const runSeed = async ({ allowBlocked, dryRun, pack, profile }: SeedOptions) => {
+  console.warn(`[seed] bootstrapping auth state against ${baseURL}`)
+  await ensureAuth()
+
+  const consoleSession = await createStandaloneConsoleSession()
+  try {
+    const results = await runSeedTasks(getTasks(pack, profile), {
+      consoleClient: consoleSession.client,
+      dryRun,
+      resources: new Map(),
+    })
+    const reportName = `${pack}-${profile}`
+    const reportPath = await writeSeedReport(reportName, results)
+    const blockedCount = results.filter((result) => result.status === 'blocked').length
+
+    console.warn(`[seed] report ${reportPath}`)
+    if (blockedCount > 0 && !allowBlocked) {
+      throw new Error(
+        `${blockedCount} seed task${blockedCount === 1 ? '' : 's'} blocked. Re-run with --allow-blocked only when partial readiness is intentional.`,
+      )
+    }
+  } finally {
+    await consoleSession.dispose()
+  }
+}

e2e/scripts/seed.ts

@@ -1,163 +0,0 @@
-import type { ManagedProcess } from '../support/process'
-import { mkdir } from 'node:fs/promises'
-import path from 'node:path'
-import { chromium } from '@playwright/test'
-import { createAgentV2SeedTasks } from '../features/agent-v2/support/seed'
-import { ensureAuthenticatedState } from '../fixtures/auth'
-import { createStandaloneConsoleSession } from '../support/api/console-session'
-import { startLoggedProcess, stopManagedProcess, waitForUrl } from '../support/process'
-import { runSeedTasks, writeSeedReport } from '../support/seed'
-import { startWebServer, stopWebServer } from '../support/web-server'
-import { apiURL, baseURL, reuseExistingWebServer } from '../test-env'
-import { e2eDir, isMainModule } from './common'
-import './env-register'
-
-type SeedOptions = {
-  allowBlocked: boolean
-  dryRun: boolean
-  pack: string
-  profile: string
-}
-
-const parseArgs = (argv: string[]): SeedOptions => {
-  const options: SeedOptions = {
-    allowBlocked: false,
-    dryRun: false,
-    pack: 'agent-v2',
-    profile: 'post-merge',
-  }
-
-  for (const [index, arg] of argv.entries()) {
-    if (arg === '--pack') {
-      options.pack = argv[index + 1] || options.pack
-      continue
-    }
-    if (arg.startsWith('--pack=')) {
-      options.pack = arg.slice('--pack='.length)
-      continue
-    }
-    if (arg === '--dry-run') options.dryRun = true
-    if (arg === '--allow-blocked') options.allowBlocked = true
-    if (arg === '--profile') {
-      options.profile = argv[index + 1] || options.profile
-      continue
-    }
-    if (arg.startsWith('--profile=')) {
-      options.profile = arg.slice('--profile='.length)
-      continue
-    }
-  }
-
-  return options
-}
-
-const getTasks = (pack: string, profile: string) => {
-  if (pack === 'agent-v2') return createAgentV2SeedTasks(profile)
-
-  throw new Error(`Unknown seed pack "${pack}".`)
-}
-
-const ensureAuth = async () => {
-  const browser = await chromium.launch({ headless: true })
-  try {
-    await ensureAuthenticatedState(browser, baseURL)
-  } finally {
-    await browser.close()
-  }
-}
-
-const startApiProcess = async (logDir: string) => {
-  try {
-    await waitForUrl(`${apiURL}/health`, 1_000, 250, 1_000)
-    return undefined
-  } catch {
-    // Start a local API process below.
-  }
-
-  const apiProcess = await startLoggedProcess({
-    command: 'npx',
-    args: ['tsx', './scripts/setup.ts', 'api'],
-    cwd: e2eDir,
-    label: 'api server',
-    logFilePath: path.join(logDir, 'seed-api.log'),
-  })
-
-  try {
-    await waitForUrl(`${apiURL}/health`, 180_000, 1_000)
-    return apiProcess
-  } catch (error) {
-    await stopManagedProcess(apiProcess)
-    throw error
-  }
-}
-
-const startCeleryProcess = async (logDir: string) =>
-  startLoggedProcess({
-    command: 'npx',
-    args: [
-      'tsx',
-      './scripts/setup.ts',
-      'celery',
-      '--queues',
-      'dataset,priority_dataset,workflow_based_app_execution',
-    ],
-    cwd: e2eDir,
-    label: 'celery worker',
-    logFilePath: path.join(logDir, 'seed-celery.log'),
-  })
-
-const main = async () => {
-  const options = parseArgs(process.argv.slice(2))
-  const logDir = path.join(e2eDir, '.logs')
-  let apiProcess: ManagedProcess | undefined
-  let celeryProcess: ManagedProcess | undefined
-  let consoleSession: Awaited<ReturnType<typeof createStandaloneConsoleSession>> | undefined
-
-  await mkdir(logDir, { recursive: true })
-
-  try {
-    apiProcess = await startApiProcess(logDir)
-    celeryProcess = await startCeleryProcess(logDir)
-    await startWebServer({
-      baseURL,
-      command: 'npx',
-      args: ['tsx', './scripts/setup.ts', 'web'],
-      cwd: e2eDir,
-      logFilePath: path.join(logDir, 'seed-web.log'),
-      reuseExistingServer: reuseExistingWebServer,
-      timeoutMs: 300_000,
-    })
-
-    console.warn(`[seed] bootstrapping auth state against ${baseURL}`)
-    await ensureAuth()
-    consoleSession = await createStandaloneConsoleSession()
-
-    const results = await runSeedTasks(getTasks(options.pack, options.profile), {
-      consoleClient: consoleSession.client,
-      dryRun: options.dryRun,
-      resources: new Map(),
-    })
-    const reportName = `${options.pack}-${options.profile}`
-    const reportPath = await writeSeedReport(reportName, results)
-    const blockedCount = results.filter((result) => result.status === 'blocked').length
-
-    console.warn(`[seed] report ${reportPath}`)
-    if (blockedCount > 0 && !options.allowBlocked) {
-      throw new Error(
-        `${blockedCount} seed task${blockedCount === 1 ? '' : 's'} blocked. Re-run with --allow-blocked only when partial readiness is intentional.`,
-      )
-    }
-  } finally {
-    await consoleSession?.dispose()
-    await stopWebServer()
-    await stopManagedProcess(celeryProcess)
-    await stopManagedProcess(apiProcess)
-  }
-}
-
-if (isMainModule(import.meta.url)) {
-  void main().catch((error) => {
-    console.error(error instanceof Error ? error.message : String(error))
-    process.exit(1)
-  })
-}

e2e/scripts/setup.ts

@@ -27,6 +27,7 @@ import {
   waitForCondition,
   webDir,
 } from './common'
+import './env-register'
 
 const buildIdPath = path.join(webDir, '.next', 'BUILD_ID')
 const webBuildStampPath = path.join(webDir, '.next', 'e2e-web-build.sha256')

e2e/test-env.ts

@@ -1,3 +1,5 @@
+import './scripts/env-register'
+
 export const defaultBaseURL = 'http://127.0.0.1:3000'
 export const defaultApiURL = 'http://127.0.0.1:5001'
 export const defaultLocale = 'en-US'

e2e/tests/run-options.test.ts

@@ -0,0 +1,65 @@
+import { describe, expect, it } from 'vitest'
+import { parseRunOptions, shouldStartManagedAgentBackend } from '../scripts/run-options'
+
+describe('E2E run options', () => {
+  it('forwards Cucumber arguments without requesting seed data', () => {
+    expect(parseRunOptions(['--tags', '@smoke'])).toEqual({
+      forwardArgs: ['--tags', '@smoke'],
+      full: false,
+      headed: false,
+      seed: undefined,
+      seedOnly: false,
+    })
+  })
+
+  it('uses the post-merge profile for the default seed command', () => {
+    expect(parseRunOptions(['--seed-only'])).toMatchObject({
+      forwardArgs: [],
+      seed: {
+        allowBlocked: false,
+        dryRun: false,
+        pack: 'agent-v2',
+        profile: 'post-merge',
+      },
+      seedOnly: true,
+    })
+  })
+
+  it('seeds a named profile before forwarding Cucumber arguments', () => {
+    expect(parseRunOptions(['--profile', 'prepared', '--', '--tags', '@prepared'])).toMatchObject({
+      forwardArgs: ['--tags', '@prepared'],
+      seed: { profile: 'prepared' },
+      seedOnly: false,
+    })
+  })
+
+  it('rejects seed-only options when no seed was requested', () => {
+    expect(() => parseRunOptions(['--allow-blocked'])).toThrow(
+      'Seed options require --seed-only or --profile.',
+    )
+    expect(() => parseRunOptions(['--dry-run', '--profile', 'prepared'])).toThrow(
+      '--dry-run requires --seed-only.',
+    )
+  })
+})
+
+describe('managed Agent backend selection', () => {
+  it.each(['1', 'true'])('starts a managed backend for %s', (value) => {
+    expect(shouldStartManagedAgentBackend({ E2E_START_AGENT_BACKEND: value })).toBe(true)
+  })
+
+  it('uses an explicitly configured backend without starting a managed one', () => {
+    expect(
+      shouldStartManagedAgentBackend({ E2E_AGENT_BACKEND_URL: 'http://agent.example.test' }),
+    ).toBe(false)
+  })
+
+  it('rejects two Agent backend owners', () => {
+    expect(() =>
+      shouldStartManagedAgentBackend({
+        AGENT_BACKEND_BASE_URL: 'http://agent.example.test',
+        E2E_START_AGENT_BACKEND: '1',
+      }),
+    ).toThrow('E2E_START_AGENT_BACKEND cannot be enabled')
+  })
+})