Compare commits

...
47 changed files with 786 additions and 589 deletions
-1
View File
@@ -63,7 +63,6 @@ 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
+2 -2
View File
@@ -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,
home_snapshot_id: str | None,
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,
home_snapshot_id: str | None,
conversation_id: str,
session_scope_snapshot_id: str | None | _DefaultSessionScopeSnapshotId,
agent_config_version_kind: AgentConfigVersionKind,
+1 -1
View File
@@ -31,7 +31,7 @@ class AgentAppSessionScope:
conversation_id: str
agent_id: str
agent_config_snapshot_id: str
home_snapshot_id: str
home_snapshot_id: str | None
agent_config_version_kind: AgentConfigVersionKind = AgentConfigVersionKind.SNAPSHOT
build_draft_id: str | None = None
@@ -125,7 +125,7 @@ class WorkflowAgentWorkspaceStore:
)
def load_or_create_node_execution_session(
self, scope: WorkflowAgentSessionScope, *, home_snapshot_id: str
self, scope: WorkflowAgentSessionScope, *, home_snapshot_id: str | None
) -> StoredWorkflowAgentSession:
with session_factory.create_session() as session:
execution = self._load_execution(session=session, scope=scope)
@@ -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=False))
batch_op.add_column(sa.Column('home_snapshot_id', models.types.StringUUID(), nullable=True))
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=False))
batch_op.add_column(sa.Column('home_snapshot_id', models.types.StringUUID(), nullable=True))
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=False))
batch_op.add_column(sa.Column('home_snapshot_id', models.types.StringUUID(), nullable=True))
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'))
@@ -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=False),
sa.Column('base_home_snapshot_id', models.types.StringUUID(), nullable=True),
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=False),
sa.Column('home_snapshot_id', sa.UUID(), autoincrement=False, nullable=True),
sa.PrimaryKeyConstraint('id', name=op.f('agent_runtime_session_pkey'))
)
with op.batch_alter_table('agent_runtime_sessions', schema=None) as batch_op:
@@ -0,0 +1,49 @@
"""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
+3 -3
View File
@@ -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] = mapped_column(StringUUID, nullable=False)
home_snapshot_id: Mapped[str | None] = mapped_column(StringUUID, nullable=True)
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] = mapped_column(StringUUID, nullable=False)
home_snapshot_id: Mapped[str | None] = mapped_column(StringUUID, nullable=True)
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] = mapped_column(StringUUID, nullable=False)
base_home_snapshot_id: Mapped[str | None] = mapped_column(StringUUID, nullable=True)
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
+4 -14
View File
@@ -510,11 +510,6 @@ 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,
@@ -523,7 +518,7 @@ class AgentComposerService:
agent_soul=AgentSoulConfig(),
operation=AgentConfigRevisionOperation.CREATE_VERSION,
version_note=None,
home_snapshot_id=home_snapshot.id,
home_snapshot_id=None,
)
agent.active_config_snapshot_id = initial_version.id
agent.active_config_has_model = False
@@ -601,7 +596,7 @@ class AgentComposerService:
tenant_id: str,
agent: Agent,
agent_soul: AgentSoulConfig,
home_snapshot_id: str,
home_snapshot_id: str | None,
) -> bool:
if not agent.active_config_snapshot_id:
return False
@@ -1767,11 +1762,6 @@ 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,
@@ -1780,7 +1770,7 @@ class AgentComposerService:
agent_soul=agent_soul,
operation=AgentConfigRevisionOperation.CREATE_VERSION,
version_note=None,
home_snapshot_id=home_snapshot.id,
home_snapshot_id=None,
)
agent.active_config_snapshot_id = version.id
agent.active_config_has_model = agent_soul_has_model(agent_soul)
@@ -1951,7 +1941,7 @@ class AgentComposerService:
agent_soul: AgentSoulConfig,
operation: AgentConfigRevisionOperation,
version_note: str | None,
home_snapshot_id: str,
home_snapshot_id: str | None,
previous_snapshot_id: str | None = None,
) -> AgentConfigSnapshot:
next_version = (
+1 -7
View File
@@ -49,7 +49,6 @@ 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
@@ -569,17 +568,12 @@ 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=home_snapshot.id,
home_snapshot_id=None,
created_by=account_id,
)
self.session.add(snapshot)
+4 -30
View File
@@ -5,7 +5,7 @@ from __future__ import annotations
import logging
from dify_agent.client import Client, DifyAgentNotFoundError
from dify_agent.protocol import CreateHomeSnapshotFromBindingRequest, InitializeHomeSnapshotRequest
from dify_agent.protocol import CreateHomeSnapshotFromBindingRequest
from sqlalchemy import select
from sqlalchemy.orm import Session
@@ -36,34 +36,6 @@ 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,
@@ -211,7 +183,9 @@ class AgentHomeSnapshotService:
return Client(base_url=base_url)
def validate_home_snapshot_binding(*, session: Session, agent: Agent, home_snapshot_id: str) -> None:
def validate_home_snapshot_binding(*, session: Session, agent: Agent, home_snapshot_id: str | None) -> None:
if home_snapshot_id is None:
return
_require_owned_home_snapshot(session=session, agent=agent, home_snapshot_id=home_snapshot_id)
+2 -12
View File
@@ -350,17 +350,12 @@ 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=home_snapshot.id,
home_snapshot_id=None,
version_note=payload.version_note,
created_by=account_id,
)
@@ -429,17 +424,12 @@ 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=home_snapshot.id,
home_snapshot_id=None,
created_by=account_id,
)
self._session.add(version)
+15 -12
View File
@@ -110,7 +110,7 @@ class AgentWorkspaceService:
session: Session,
scope: WorkspaceOwnerScope,
agent_id: str,
base_home_snapshot_id: str,
base_home_snapshot_id: str | None,
agent_config_version_id: str,
agent_config_version_kind: AgentConfigVersionKind,
) -> AgentWorkspaceBinding:
@@ -123,16 +123,19 @@ class AgentWorkspaceService:
fails before the backend returns success.
"""
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_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,
)
)
)
if home_snapshot is None:
raise AgentWorkspaceNotFoundError("base Home Snapshot is unavailable")
if home_snapshot is None:
raise AgentWorkspaceNotFoundError("base Home Snapshot is unavailable")
home_snapshot_ref = home_snapshot.snapshot_ref
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())
@@ -144,7 +147,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.snapshot_ref,
home_snapshot_ref=home_snapshot_ref,
)
)
if workspace is not None and allocation.workspace_ref != workspace.backend_workspace_ref:
@@ -439,7 +442,7 @@ class AgentWorkspaceService:
def validate_binding_generation(
binding: AgentWorkspaceBinding,
*,
base_home_snapshot_id: str,
base_home_snapshot_id: str | None,
agent_config_version_id: str,
agent_config_version_kind: AgentConfigVersionKind,
) -> None:
@@ -124,7 +124,6 @@ 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()
@@ -143,10 +142,6 @@ 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,
@@ -1,5 +1,4 @@
from datetime import datetime
from types import SimpleNamespace
from unittest.mock import create_autospec, patch
import pytest
@@ -30,7 +29,6 @@ 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
@@ -44,10 +42,6 @@ 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,
@@ -16,6 +16,7 @@ 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",
@@ -23,7 +24,7 @@ def _scope(
conversation_id="conversation-1",
agent_id="agent-1",
agent_config_snapshot_id="config-1",
home_snapshot_id="home-1",
home_snapshot_id=home_snapshot_id,
agent_config_version_kind=kind,
build_draft_id=build_draft_id,
)
@@ -54,7 +55,8 @@ def test_scope_selects_conversation_or_build_draft_workspace_owner() -> None:
assert build_owner.owner_id == "build-draft-1"
def test_load_or_create_persists_new_binding_on_caller(monkeypatch) -> None:
@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:
caller = SimpleNamespace(agent_workspace_binding_id=None)
context = MagicMock()
session = context.__enter__.return_value
@@ -64,13 +66,14 @@ def test_load_or_create_persists_new_binding_on_caller(monkeypatch) -> None:
monkeypatch.setattr(store, "_load_caller", MagicMock(return_value=caller))
monkeypatch.setattr(AgentWorkspaceService, "create_binding", create)
stored = store.load_or_create(_scope())
stored = store.load_or_create(_scope(home_snapshot_id=home_snapshot_id))
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()
@@ -155,7 +155,8 @@ def test_load_existing_scope_rejects_unavailable_persisted_binding(monkeypatch:
)
def test_load_or_create_persists_binding_on_node_execution(monkeypatch) -> None:
@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:
execution = WorkflowNodeExecutionModel(
agent_workspace_binding_id=None,
process_data=json.dumps({"existing": "value"}),
@@ -171,7 +172,7 @@ def test_load_or_create_persists_binding_on_node_execution(monkeypatch) -> None:
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-1")
stored = store.load_or_create_node_execution_session(_scope(), home_snapshot_id=home_snapshot_id)
assert stored.binding_id == "binding-1"
assert stored.workspace_id == "workspace-1"
@@ -183,6 +184,7 @@ def test_load_or_create_persists_binding_on_node_execution(monkeypatch) -> None:
}
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())
@@ -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=False),
sa.Column("home_snapshot_id", sa.String(36), nullable=True),
sa.Column("backend_run_id", sa.String(255)),
sa.Column("status", sa.String(32), nullable=False),
)
@@ -110,7 +110,10 @@ 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_columns = {column["name"] for column in inspector.get_columns("agent_workspace_bindings")}
binding_column_definitions = {
column["name"]: column for column in inspector.get_columns("agent_workspace_bindings")
}
binding_columns = set(binding_column_definitions)
assert {
"workspace_id",
"agent_id",
@@ -125,6 +128,7 @@ 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")}
@@ -0,0 +1,181 @@
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"
)
@@ -684,12 +684,10 @@ def test_resolve_package_soul_preserves_existing_and_marks_missing_knowledge(mon
}
def test_create_snapshot_increments_version_and_records_revision(monkeypatch: pytest.MonkeyPatch) -> None:
def test_create_snapshot_increments_version_and_records_revision() -> 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",
@@ -700,12 +698,7 @@ def test_create_snapshot_increments_version_and_records_revision(monkeypatch: py
)
assert snapshot.version == 3
assert snapshot.home_snapshot_id == "home-3"
create_initial.assert_called_once_with(
session=session,
tenant_id="tenant-1",
agent_id="agent-1",
)
assert snapshot.home_snapshot_id is None
assert isinstance(session.add.call_args_list[0].args[0], AgentConfigSnapshot)
revision = session.add.call_args_list[1].args[0]
assert isinstance(revision, AgentConfigRevision)
@@ -129,15 +129,6 @@ 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(
{
@@ -612,7 +603,6 @@ 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(
@@ -646,7 +636,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 == "home-initial"
assert fake_session.added[1].home_snapshot_id is None
assert fake_session.added[0].active_config_is_published is False
assert fake_session.flushes >= 1
@@ -869,7 +859,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="home-1",
home_snapshot_id=None,
config_snapshot=_agent_soul_with_model(),
)
version = SimpleNamespace(id="version-2")
@@ -905,7 +895,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"] == "home-1"
assert created["home_snapshot_id"] is None
assert calls == ["validate_home", "create_version"]
assert agent.active_config_snapshot_id == "version-2"
assert agent.active_config_has_model is True
@@ -939,7 +929,6 @@ 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)
@@ -951,7 +940,6 @@ 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(
@@ -971,7 +959,6 @@ 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()
@@ -1233,7 +1220,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="home-old",
home_snapshot_id=None,
agent_workspace_binding_id="binding-1",
config_snapshot=AgentSoulConfig(),
)
@@ -1243,14 +1230,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="home-old",
home_snapshot_id=None,
config_snapshot=AgentSoulConfig(),
)
source_binding = SimpleNamespace(
id="binding-1",
backend_binding_ref="backend-binding-1",
agent_id="agent-1",
base_home_snapshot_id="home-old",
base_home_snapshot_id=None,
agent_config_version_id="build-1",
agent_config_version_kind=AgentConfigVersionKind.BUILD_DRAFT,
)
@@ -1762,10 +1749,8 @@ 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)
@@ -1791,7 +1776,6 @@ 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()
@@ -3168,7 +3152,6 @@ 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",
@@ -3199,7 +3182,6 @@ 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,
@@ -3208,14 +3190,11 @@ 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="home-roster-initial",
home_snapshot_id=None,
),
)
monkeypatch.setattr(
AgentComposerService,
"_create_config_version",
lambda **kwargs: SimpleNamespace(id="version-with-model"),
)
create_config_version = MagicMock(return_value=SimpleNamespace(id="version-with-model"))
monkeypatch.setattr(AgentComposerService, "_create_config_version", create_config_version)
workflow_agent = AgentComposerService._create_workflow_only_agent(
session=session,
@@ -3239,11 +3218,8 @@ 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"
create_initial.assert_called_once_with(
session=session,
tenant_id="tenant-1",
agent_id=workflow_agent.id,
)
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)
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
@@ -3962,13 +3938,6 @@ 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",
@@ -4006,10 +3975,9 @@ 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)
] == ["home-roster", "home-backing"]
] == [None, None]
assert created.active_config_has_model is False
assert backing_agent.role == "Support agent"
assert backing_agent.active_config_snapshot_id is not None
@@ -4553,9 +4521,6 @@ 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",
@@ -4577,12 +4542,7 @@ 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 == "home-1"
create_initial.assert_called_once_with(
session=session,
tenant_id="tenant-1",
agent_id=agent.id,
)
assert snapshots[0].home_snapshot_id is None
assert agent.active_config_snapshot_id == snapshots[0].id
revisions = [
a for a in session.added if getattr(a, "operation", None) == AgentConfigRevisionOperation.CREATE_VERSION
@@ -4966,7 +4926,6 @@ 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:
@@ -4992,7 +4951,6 @@ 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",
@@ -5027,16 +4985,11 @@ 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 == "home-target"
assert target_version.home_snapshot_id is None
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):
@@ -6,6 +6,7 @@ import pytest
from sqlalchemy.orm import Session
from models.agent import (
Agent,
AgentConfigDraft,
AgentConfigDraftType,
AgentConfigSnapshot,
@@ -14,11 +15,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
from services.agent.home_snapshot_service import AgentHomeSnapshotService, validate_home_snapshot_binding
from services.agent.workspace_service import AgentWorkspaceService
def _build_draft() -> AgentConfigDraft:
def _build_draft(*, home_snapshot_id: str | None = "home-old") -> AgentConfigDraft:
return AgentConfigDraft(
id="build-1",
tenant_id="tenant-1",
@@ -26,7 +27,7 @@ def _build_draft() -> AgentConfigDraft:
draft_type=AgentConfigDraftType.DEBUG_BUILD,
account_id="account-1",
draft_owner_key="account-1",
home_snapshot_id="home-old",
home_snapshot_id=home_snapshot_id,
agent_workspace_binding_id="binding-1",
config_snapshot=AgentSoulConfig(),
)
@@ -34,42 +35,19 @@ def _build_draft() -> AgentConfigDraft:
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_create_initial_persists_backend_snapshot_ref(monkeypatch: pytest.MonkeyPatch) -> None:
def test_validate_home_snapshot_binding_accepts_default_home_without_ledger_lookup() -> None:
session = MagicMock()
client = _client()
monkeypatch.setattr(AgentHomeSnapshotService, "_client", lambda: nullcontext(client))
snapshot = AgentHomeSnapshotService.create_initial(
validate_home_snapshot_binding(
session=session,
tenant_id="tenant-1",
agent_id="agent-1",
agent=Agent(id="agent-1"),
home_snapshot_id=None,
)
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()
session.scalar.assert_not_called()
@pytest.mark.parametrize(
@@ -98,7 +76,8 @@ 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)
monkeypatch.setattr(AgentWorkspaceService, "validate_binding_generation", MagicMock())
validate_generation = MagicMock()
monkeypatch.setattr(AgentWorkspaceService, "validate_binding_generation", validate_generation)
snapshot = AgentHomeSnapshotService.create_for_build_apply(
session=session,
@@ -110,6 +89,32 @@ 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:
@@ -15,7 +15,7 @@ from models.agent import (
AgentWorkspaceBinding,
AgentWorkspaceOwnerType,
)
from services.agent.workspace_service import AgentWorkspaceService, WorkspaceOwnerScope
from services.agent.workspace_service import AgentWorkspaceNotFoundError, AgentWorkspaceService, WorkspaceOwnerScope
def _scope() -> WorkspaceOwnerScope:
@@ -127,6 +127,59 @@ 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(
@@ -67,12 +67,17 @@ streams are observability state, not the Home/Workspace/Binding ledger.
## Creation and execution flow
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
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
when the source Binding is unavailable.
Before an Agent request, Dify API loads the specific product context. If it has
@@ -82,10 +87,12 @@ 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` 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:
`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:
```json
{
@@ -174,9 +181,9 @@ own Home plus the shared Workspace.
| Backend | Home Snapshot operations | Binding operations | Physical relationship |
| --- | --- | --- | --- |
| 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 | 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 creates a new Home for every Binding id. Destroying one Binding without
the Workspace leaves sibling Homes and the shared Workspace intact. Current E2B
@@ -185,9 +192,8 @@ 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; temporary Home initialization
resources are killed. It is not a retention TTL and does not delete paused
resources or immutable snapshots.
resource. Runtime resources pause on timeout. 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
@@ -85,10 +85,9 @@ 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 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`.
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.
A shell-enabled request includes Execution Context, `dify.runtime`, and
`dify.shell`. Dify API creates or resolves the specific persistent Binding for
+6 -9
View File
@@ -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. Current Home Snapshot and Binding operations fail fast with `NotImplementedError`. |
| `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_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 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_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_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,12 +177,9 @@ docker compose \
```
`DIFY_AGENT_E2B_ACTIVE_TIMEOUT_SECONDS` controls continuous active E2B time.
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.
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.
## Run runtime-backend integration contracts
@@ -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 settings
can be selected, but resource operations currently fail fast with
`NotImplementedError`; there is no compatibility fallback to the retired
Sandbox protocol.
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.
```python
from dify_agent.runtime.compositor_factory import create_default_layer_providers
@@ -36,7 +36,6 @@ from dify_agent.protocol import (
DeleteHomeSnapshotRequest,
DestroyExecutionBindingRequest,
HomeSnapshotResponse,
InitializeHomeSnapshotRequest,
RUN_EVENT_ADAPTER,
RunEvent,
RunEventsResponse,
@@ -529,24 +528,6 @@ 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,
@@ -46,7 +46,6 @@ from .home_snapshot import (
CreateHomeSnapshotFromBindingRequest,
DeleteHomeSnapshotRequest,
HomeSnapshotResponse,
InitializeHomeSnapshotRequest,
)
from .workspace import (
WorkspaceFileEntry,
@@ -79,7 +78,6 @@ __all__ = [
"EmptyRunEventData",
"LayerExitSignals",
"HomeSnapshotResponse",
"InitializeHomeSnapshotRequest",
"PydanticAIStreamRunEvent",
"RUN_EVENT_ADAPTER",
"RunCancelledEvent",
@@ -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 = Field(min_length=1)
home_snapshot_ref: str | None = Field(default=None, min_length=1)
model_config: ClassVar[ConfigDict] = ConfigDict(extra="forbid")
@@ -5,14 +5,6 @@ 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)
@@ -38,5 +30,4 @@ __all__ = [
"CreateHomeSnapshotFromBindingRequest",
"DeleteHomeSnapshotRequest",
"HomeSnapshotResponse",
"InitializeHomeSnapshotRequest",
]
@@ -22,7 +22,6 @@ from .protocols import (
FileSystem,
HomeSnapshotBackend,
HomeSnapshotCreateSpec,
InitializeHomeSnapshotSpec,
RuntimeBackendProfile,
RuntimeLayout,
RuntimeLease,
@@ -46,7 +45,6 @@ __all__ = [
"HomeSnapshotCreateError",
"HomeSnapshotCreateSpec",
"HomeSnapshotNotFoundError",
"InitializeHomeSnapshotSpec",
"RuntimeBackendError",
"RuntimeBackendProfile",
"RuntimeLayout",
@@ -30,7 +30,6 @@ from dify_agent.runtime_backend.protocols import (
ExecutionBindingDestroySpec,
FileSystem,
HomeSnapshotCreateSpec,
InitializeHomeSnapshotSpec,
RuntimeLayout,
RuntimeLease,
)
@@ -163,44 +162,12 @@ class E2BSDKControlPlane:
class E2BHomeSnapshotBackend:
"""Implement immutable Home Snapshot operations with E2B snapshots.
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.
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."""
@@ -236,6 +203,7 @@ class E2BExecutionBindingBackend:
"""
control_plane: E2BControlPlane
template: str
active_timeout_seconds: int
shellctl_auth_token: str = ""
shellctl_port: int = 5004
@@ -244,13 +212,13 @@ class E2BExecutionBindingBackend:
)
async def create_binding(self, spec: ExecutionBindingCreateSpec) -> ExecutionBindingAllocation:
"""Create one paused E2B resource from an immutable Home Snapshot ref."""
"""Create one paused E2B resource from a snapshot or deployment template."""
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(
spec.home_snapshot_ref,
self.template if spec.home_snapshot_ref is None else spec.home_snapshot_ref,
timeout=self.active_timeout_seconds,
metadata={
"dify.resource": "runtime-sandbox",
@@ -274,7 +242,7 @@ class E2BExecutionBindingBackend:
except BaseException:
pass
if isinstance(exc, Exception):
if isinstance(exc, (BindingCreateError, SharedWorkspaceUnsupportedError)):
if isinstance(exc, BindingCreateError):
raise
raise BindingCreateError(str(exc)) from exc
raise
@@ -1,10 +1,10 @@
"""Enterprise Gateway adapter for the working-environment protocol.
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.
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.
"""
from __future__ import annotations
@@ -21,8 +21,10 @@ 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 (
@@ -31,7 +33,6 @@ from dify_agent.runtime_backend.protocols import (
ExecutionBindingDestroySpec,
FileSystem,
HomeSnapshotCreateSpec,
InitializeHomeSnapshotSpec,
RuntimeLayout,
RuntimeLease,
)
@@ -45,21 +46,13 @@ logger = logging.getLogger(__name__)
def _not_implemented() -> NotImplementedError:
return NotImplementedError("Enterprise Gateway does not implement the Execution Binding protocol")
return NotImplementedError("Enterprise Gateway does not implement immutable Home Snapshot operations")
@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()
@@ -71,7 +64,7 @@ class EnterpriseHomeSnapshotBackend:
@dataclass(slots=True)
class EnterpriseExecutionBindingBackend:
"""Access and destroy legacy Gateway sandboxes as coupled physical Bindings."""
"""Manage Gateway sandboxes as coupled physical Bindings and Workspaces."""
gateway_endpoint: str
auth_token: str
@@ -82,8 +75,56 @@ class EnterpriseExecutionBindingBackend:
)
async def create_binding(self, spec: ExecutionBindingCreateSpec) -> ExecutionBindingAllocation:
del spec
raise _not_implemented()
"""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
async def acquire(self, binding_ref: str) -> RuntimeLease:
"""Reconnect to one existing Gateway sandbox without creating a replacement."""
@@ -137,21 +178,34 @@ 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:
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()
await self._delete_sandbox(spec.binding_ref)
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}
@@ -27,7 +27,6 @@ from dify_agent.runtime_backend.protocols import (
ExecutionBindingCreateSpec,
ExecutionBindingDestroySpec,
HomeSnapshotCreateSpec,
InitializeHomeSnapshotSpec,
RuntimeLayout,
RuntimeLease,
)
@@ -49,28 +48,6 @@ 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)
@@ -142,31 +119,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)}"
)
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)}",
]
)
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)
try:
result = await run_shellctl_control_command(lease.commands, script)
if result.exit_code != 0:
@@ -99,11 +99,7 @@ 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(
gateway_endpoint=endpoint,
auth_token=token,
gateway_timeout=settings.enterprise_sandbox_gateway_timeout,
),
home_snapshots=EnterpriseHomeSnapshotBackend(),
execution_bindings=EnterpriseExecutionBindingBackend(
gateway_endpoint=endpoint,
auth_token=token,
@@ -116,11 +112,10 @@ 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,
@@ -13,13 +13,6 @@ 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
@@ -100,7 +93,7 @@ class ExecutionBindingCreateSpec:
binding_id: str
workspace_id: str
existing_workspace_ref: str | None
home_snapshot_ref: str
home_snapshot_ref: str | None = None
@dataclass(frozen=True, slots=True)
@@ -126,12 +119,18 @@ class ExecutionBindingBackend(Protocol):
async def create_binding(self, spec: ExecutionBindingCreateSpec) -> ExecutionBindingAllocation:
"""Materialize a mutable Home and make the requested Workspace ready.
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.
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.
"""
...
@@ -171,16 +170,6 @@ 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.
@@ -217,7 +206,6 @@ __all__ = [
"FileSystem",
"HomeSnapshotBackend",
"HomeSnapshotCreateSpec",
"InitializeHomeSnapshotSpec",
"RuntimeBackendProfile",
"RuntimeLayout",
"RuntimeLease",
@@ -6,7 +6,6 @@ from dify_agent.protocol.home_snapshot import (
CreateHomeSnapshotFromBindingRequest,
DeleteHomeSnapshotRequest,
HomeSnapshotResponse,
InitializeHomeSnapshotRequest,
)
from dify_agent.runtime_backend import (
BindingAcquireError,
@@ -16,7 +15,6 @@ from dify_agent.runtime_backend import (
HomeSnapshotCreateError,
HomeSnapshotCreateSpec,
HomeSnapshotNotFoundError,
InitializeHomeSnapshotSpec,
)
from dify_agent.runtime_backend.leases import open_runtime_lease
@@ -38,19 +36,6 @@ 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,
@@ -9,7 +9,6 @@ from dify_agent.protocol.home_snapshot import (
CreateHomeSnapshotFromBindingRequest,
DeleteHomeSnapshotRequest,
HomeSnapshotResponse,
InitializeHomeSnapshotRequest,
)
from dify_agent.server.home_snapshots import HomeSnapshotService, HomeSnapshotServiceError
@@ -26,19 +25,6 @@ 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,
@@ -12,10 +12,9 @@ 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, LocalHomeSnapshotBackend
from dify_agent.runtime_backend.local import LocalExecutionBindingBackend
pytestmark = pytest.mark.integration
@@ -32,19 +31,10 @@ 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",
@@ -52,7 +42,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=snapshot_ref,
home_snapshot_ref=None,
)
)
allocations.append(first)
@@ -71,7 +61,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=snapshot_ref,
home_snapshot_ref=None,
)
)
allocations.append(second)
@@ -102,11 +92,6 @@ 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]
@@ -120,22 +105,14 @@ async def test_e2b_binding_checkpoint_and_collection() -> None:
)
marker = uuid.uuid4().hex
control = E2BSDKControlPlane(api_key=api_key)
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
snapshots = E2BHomeSnapshotBackend(control_plane=control)
bindings = E2BExecutionBindingBackend(control_plane=control, template=template, active_timeout_seconds=3600)
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",
@@ -143,7 +120,7 @@ async def test_e2b_binding_checkpoint_and_collection() -> None:
binding_id=marker,
workspace_id=marker,
existing_workspace_ref=None,
home_snapshot_ref=snapshot_ref,
home_snapshot_ref=None,
)
)
lease = await bindings.acquire(allocation.binding_ref)
@@ -211,11 +188,10 @@ async def test_e2b_binding_checkpoint_and_collection() -> None:
)
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 checkpoint_ref is not None:
try:
await snapshots.delete(checkpoint_ref)
except BaseException as exc:
cleanup_errors.append(exc)
if cleanup_errors and not primary_error:
raise cleanup_errors[0]
@@ -28,7 +28,6 @@ from dify_agent.protocol import (
CreateHomeSnapshotFromBindingRequest,
CreateRunRequest,
DestroyExecutionBindingRequest,
InitializeHomeSnapshotRequest,
RUN_EVENT_ADAPTER,
RunCancelledEvent,
RunEvent,
@@ -311,14 +310,6 @@ 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))
@@ -368,12 +359,9 @@ def _create_home_snapshot_from_binding_request() -> CreateHomeSnapshotFromBindin
)
def test_sync_home_snapshot_client_parses_initialize_checkpoint_and_delete() -> None:
def test_sync_home_snapshot_client_parses_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"
@@ -387,20 +375,16 @@ def test_sync_home_snapshot_client_parses_initialize_checkpoint_and_delete() ->
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_initialize_checkpoint_and_delete() -> None:
def test_async_home_snapshot_client_parses_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"
@@ -411,11 +395,9 @@ def test_async_home_snapshot_client_parses_initialize_checkpoint_and_delete() ->
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()
@@ -430,7 +412,7 @@ def test_home_snapshot_client_maps_sync_validation_and_async_http_errors() -> No
)
with pytest.raises(DifyAgentValidationError):
_ = sync_client.initialize_home_snapshot_sync(_initialize_home_snapshot_request())
_ = sync_client.create_home_snapshot_from_binding_sync(_create_home_snapshot_from_binding_request())
sync_http_client.close()
async def scenario() -> None:
@@ -29,6 +29,29 @@ 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)
@@ -10,7 +10,6 @@ from dify_agent.runtime_backend import (
ExecutionBindingCreateSpec,
ExecutionBindingDestroySpec,
HomeSnapshotCreateSpec,
InitializeHomeSnapshotSpec,
SharedWorkspaceUnsupportedError,
WorkspacePreservationUnsupportedError,
)
@@ -86,7 +85,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"] in {"home-snapshot-initialize", "runtime-sandbox"}
assert metadata["dify.resource"] == "runtime-sandbox"
return sandbox
async def connect(self, handle: str, *, timeout: int) -> _Sandbox:
@@ -103,49 +102,57 @@ class _ControlPlane:
@pytest.mark.anyio
async def test_e2b_profile_uses_snapshot_as_runtime_template_and_couples_refs() -> None:
async def test_e2b_binding_uses_default_template_or_exact_snapshot_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,
)
snapshot_ref = await snapshots.initialize(
InitializeHomeSnapshotSpec(tenant_id="tenant-1", agent_id="agent-1", home_snapshot_id="home-1")
)
allocation = await bindings.create_binding(
default_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=snapshot_ref,
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",
)
)
assert control.created == [("prepared-template", "kill"), (snapshot_ref, "pause")]
assert allocation.binding_ref == allocation.workspace_ref
runtime = control.sandboxes[allocation.binding_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 runtime.files.paths == {"/home/dify/workspace"}
assert runtime.pauses == [True]
await bindings.destroy_binding(
ExecutionBindingDestroySpec(
binding_ref=allocation.binding_ref,
workspace_ref=allocation.workspace_ref,
destroy_workspace=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 snapshots.delete(snapshot_ref)
await snapshots.delete("snapshot-1")
assert control.killed == [allocation.binding_ref]
assert control.deleted_snapshots == [snapshot_ref]
assert control.killed == [default_allocation.binding_ref, snapshot_allocation.binding_ref]
assert control.deleted_snapshots == ["snapshot-1"]
@pytest.mark.anyio
@@ -153,6 +160,7 @@ 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(
@@ -177,6 +185,7 @@ 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,
)
@@ -196,6 +205,36 @@ 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()
@@ -206,8 +245,6 @@ 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(
@@ -9,13 +9,14 @@ 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 (
@@ -282,23 +283,55 @@ async def test_enterprise_destroy_propagates_gateway_failure(
@pytest.mark.anyio
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")
async def test_enterprise_default_binding_creates_gateway_sandbox_and_layout(
monkeypatch: pytest.MonkeyPatch,
) -> None:
requests: list[httpx.Request] = []
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")
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.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(
)
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(
ExecutionBindingCreateSpec(
tenant_id="tenant-1",
agent_id="agent-1",
@@ -308,3 +341,63 @@ async def test_enterprise_allocation_and_home_snapshots_remain_explicitly_not_im
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")
@@ -13,8 +13,6 @@ from dify_agent.runtime_backend import (
ExecutionBindingCreateSpec,
ExecutionBindingDestroySpec,
HomeSnapshotCreateSpec,
HomeSnapshotCreateError,
InitializeHomeSnapshotSpec,
)
from dify_agent.runtime_backend.local import LocalExecutionBindingBackend, LocalHomeSnapshotBackend
@@ -137,42 +135,6 @@ 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()
@@ -198,13 +160,41 @@ 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 ("test", "-d", "/snapshots/home-home-1") in factory.commands
assert factory.commands[0] == ("test", "-d", "/snapshots/home-home-1")
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()
@@ -8,22 +8,16 @@ import pytest
from dify_agent.protocol import (
CreateHomeSnapshotFromBindingRequest,
DeleteHomeSnapshotRequest,
InitializeHomeSnapshotRequest,
)
from dify_agent.runtime_backend import HomeSnapshotCreateSpec, InitializeHomeSnapshotSpec, RuntimeLease
from dify_agent.runtime_backend import HomeSnapshotCreateSpec, 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"
@@ -47,7 +41,7 @@ class _BindingBackend:
@pytest.mark.anyio
async def test_home_snapshot_service_initializes_and_checkpoints_exact_binding() -> None:
async def test_home_snapshot_service_checkpoints_exact_binding() -> None:
lease = cast(RuntimeLease, object())
homes = _HomeBackend()
bindings = _BindingBackend(lease=lease)
@@ -56,9 +50,6 @@ async def test_home_snapshot_service_initializes_and_checkpoints_exact_binding()
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",
@@ -68,7 +59,6 @@ async def test_home_snapshot_service_initializes_and_checkpoints_exact_binding()
)
)
assert initial.snapshot_ref == "snapshot-initial"
assert checkpoint.snapshot_ref == "snapshot-build"
assert bindings.acquired == ["binding-ref"]
assert bindings.released == [lease]
@@ -301,6 +301,7 @@ 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: