fix(api): validate pipeline and bot writes before persistence

Co-authored-by: multica-agent <[email protected]>

skills/skills/langbot-mcp-ops/SKILL.md

@@ -60,6 +60,12 @@ To inspect key identity and permissions, call `GET /api/v1/system/context` with
 
 ## Tool surface
 
+`create_bot` requires `bot_data.name` and `bot_data.adapter` as non-empty strings,
+and `bot_data.adapter_config` as an object with the adapter's settings.
+For `update_pipeline`, fetch the current pipeline before editing: `config` is a
+complete replacement containing `trigger`, `safety`, `ai`, and `output` objects,
+not a partial patch. Invalid configurations are rejected before persistence.
+
 Slack quick setup: create a disabled `slack-omni` bot draft first, then use
 `start_slack_setup` with the user's App Configuration access/refresh tokens.
 Choose `socket_mode=true` with a browser-reachable OAuth `redirect_url`, or

src/langbot/pkg/api/application/operation_docs.py

@@ -245,7 +245,7 @@ OPERATION_DOCS: dict[str, OperationDoc] = {
         domain='bot',
         label={'zh_Hans': '创建机器人', 'en_US': 'Create Bot'},
         summary='Create a new chat bot connected to an IM adapter.',
-        description='Create and configure a new chat bot in the workspace. Provide bot_data containing name, adapter settings, and optional pipeline binding. Side effect: Instantiates a new bot entity.',
+        description='Create a chat bot. bot_data requires non-empty name and adapter strings, and an adapter_config object with adapter-specific settings. Optional event_bindings routes events to a pipeline or agent. Side effect: Instantiates a new bot entity.',
         risk='write',
         exposure='all',
     ),
@@ -997,7 +997,7 @@ OPERATION_DOCS: dict[str, OperationDoc] = {
         domain='pipeline',
         label={'zh_Hans': '更新流水线', 'en_US': 'Update Pipeline'},
         summary='Update topology, nodes, or settings of a pipeline.',
-        description='Modify pipeline nodes, model assignments, tool attachments, or name. Prerequisites: Call pipeline.list to obtain pipeline_uuid. Side effect: Updates pipeline definition.',
+        description='Modify pipeline settings or name. Call pipeline.get first: config, when supplied, replaces the complete configuration and must contain trigger, safety, ai, and output objects. Partial config is rejected without writing. Runtime initialization is checked before persistence. Side effect: Updates pipeline definition.',
         risk='write',
         exposure='all',
     ),

src/langbot/pkg/api/application/service_operations.py

@@ -834,6 +834,15 @@ def _contract(operation_id, method):
             annotation = annotation | None
         default = ... if default is inspect.Parameter.empty else default
         constraints = {}
+        if operation_id == 'bot.create' and name == 'bot_data':
+            constraints['json_schema_extra'] = {
+                'required': ['name', 'adapter', 'adapter_config'],
+                'properties': {
+                    'name': {'type': 'string', 'minLength': 1},
+                    'adapter': {'type': 'string', 'minLength': 1},
+                    'adapter_config': {'type': 'object'},
+                },
+            }
         if name in {'limit', 'max_count', 'max_entries'}:
             constraints = {'ge': 1, 'le': 100_000 if '.export.' in operation_id else 1000}
         elif name in {'offset', 'from_index', 'after_sequence', 'before_id'}:

src/langbot/pkg/api/http/service/bot.py

@@ -759,6 +759,13 @@ class BotService:
     async def create_bot(self, context: TenantContext, bot_data: dict) -> str:
         """Create bot"""
         workspace_uuid = require_workspace_uuid(context)
+        if not isinstance(bot_data, dict):
+            raise ValueError('bot_data must be an object')
+        for field in ('name', 'adapter'):
+            if not isinstance(bot_data.get(field), str) or not bot_data[field].strip():
+                raise ValueError(f'bot_data.{field} is required and must be a non-empty string')
+        if not isinstance(bot_data.get('adapter_config'), dict):
+            raise ValueError('bot_data.adapter_config is required and must be an object')
         # Check limitation
         from .workspace_limits import workspace_limits
 
@@ -769,7 +776,6 @@ class BotService:
             if len(existing_bots) >= max_bots:
                 raise ValueError(f'Maximum number of bots ({max_bots}) reached')
 
-        # TODO: 检查配置信息格式
         bot_data = await self._prepare_bot_data(context, bot_data, include_uuid=True)
         await require_workspace_platform(self.ap, workspace_uuid, bot_data['adapter'])
         bot_data['uuid'] = str(uuid.uuid4())

src/langbot/pkg/api/http/service/pipeline.py

@@ -4,6 +4,7 @@ import uuid
 import json
 import sqlalchemy
 import typing
+import copy
 
 from ....core import app
 from ....agent.runner.config_resolver import RunnerConfigResolver
@@ -229,10 +230,16 @@ class PipelineService:
         for protected_field in ('uuid', 'workspace_uuid', 'for_version', 'stages', 'is_default'):
             pipeline_data.pop(protected_field, None)
 
+        current_pipeline = await self.get_pipeline(context, pipeline_uuid, include_secret=True)
+        if current_pipeline is None:
+            raise WorkspaceNotFoundError('Pipeline not found')
         if 'config' in pipeline_data:
-            current_pipeline = await self.get_pipeline(context, pipeline_uuid, include_secret=True)
-            if current_pipeline is None:
-                raise WorkspaceNotFoundError('Pipeline not found')
+            config = pipeline_data['config']
+            if not isinstance(config, dict):
+                raise ValueError('config must be an object containing trigger, safety, ai, and output')
+            for section in ('trigger', 'safety', 'ai', 'output'):
+                if not isinstance(config.get(section), dict):
+                    raise ValueError(f'config.{section} must be an object; config replaces the complete configuration')
             current_config = current_pipeline.get('config', {})
             old_ai = current_config.get('ai', {}) if isinstance(current_config, dict) else {}
             old_runner = old_ai.get('runner') if isinstance(old_ai, dict) else None
@@ -268,6 +275,13 @@ class PipelineService:
                     new_prefs['workspace_operations'] = old_ops
             self._validate_extension_preferences(pipeline_data['extensions_preferences'])
 
+        candidate_data = copy.deepcopy(current_pipeline)
+        candidate_data.update(copy.deepcopy(pipeline_data))
+        try:
+            candidate = await self.ap.pipeline_mgr.prepare_pipeline(context, candidate_data)
+        except (KeyError, TypeError, ValueError) as exc:
+            raise ValueError('Invalid pipeline configuration: runtime initialization failed') from exc
+
         result = await self.ap.persistence_mgr.execute_async(
             scope_statement(
                 sqlalchemy.update(persistence_pipeline.LegacyPipeline)
@@ -280,12 +294,7 @@ class PipelineService:
         if getattr(result, 'rowcount', None) == 0:
             raise WorkspaceNotFoundError('Pipeline not found')
 
-        pipeline = await self.get_pipeline(context, pipeline_uuid, include_secret=True)
-        if pipeline is None:
-            raise WorkspaceNotFoundError('Pipeline not found')
-
-        await self.ap.pipeline_mgr.remove_pipeline(context, pipeline_uuid)
-        await self.ap.pipeline_mgr.load_pipeline(context, pipeline)
+        self.ap.pipeline_mgr.publish_pipeline(candidate)
 
         # update all conversation that use this pipeline
         for session in self.ap.sess_mgr.session_list:

src/langbot/pkg/api/mcp/server.py

@@ -357,7 +357,8 @@ class LangBotMCPServer:
         @mcp.tool(
             description=(
                 'Create a bot. Cloud Workspace platform restrictions also apply to this tool. `bot_data` is a JSON object matching the LangBot '
-                'POST /api/v1/platform/bots body (e.g. name, adapter, config). '
+                'POST /api/v1/platform/bots body: name and adapter are required non-empty strings; '
+                'adapter_config is a required object containing adapter-specific settings. '
                 'Use event_bindings for exclusive Agent/Pipeline routes; plugin_processors is an array '
                 'of {processor_uuid, enabled} subscriptions that automatically match Runner-declared events. '
                 'Subscriptions run independently of each other and of event_bindings. Returns the new bot UUID.'
@@ -449,7 +450,9 @@ class LangBotMCPServer:
                 }
             )
 
-        @mcp.tool(description='Update a pipeline by UUID. `pipeline_data` matches the PUT body.')
+        @mcp.tool(
+            description='Update a pipeline by UUID. `pipeline_data` matches the PUT body. Fetch the pipeline first: config replaces the complete configuration and requires trigger, safety, ai, and output objects. Invalid config is rejected before writing.'
+        )
         async def update_pipeline(pipeline_uuid: str, pipeline_data: dict) -> str:
             context = _authorized(Permission.RESOURCE_MANAGE)
             await ap.pipeline_service.update_pipeline(context, pipeline_uuid, pipeline_data)

tests/integration/api/test_pipeline_bot_write_validation.py

@@ -0,0 +1,312 @@
+import copy
+import asyncio
+import json
+import logging
+import socket
+from types import SimpleNamespace
+from unittest.mock import AsyncMock
+
+import pytest
+import quart
+import sqlalchemy as sa
+import aiohttp
+from hypercorn.asyncio import serve
+from hypercorn.config import Config
+from sqlalchemy.ext.asyncio import create_async_engine
+
+from langbot.pkg.api.application import ApplicationAPI, ApplicationAPIException
+from langbot.pkg.api.http.controller.groups.pipelines.pipelines import PipelinesRouterGroup
+from langbot.pkg.api.http.controller.groups.platform.bots import BotsRouterGroup
+from langbot.pkg.api.http.service.bot import BotService
+from langbot.pkg.api.http.service.pipeline import PipelineService
+from langbot.pkg.entity.persistence.bot import Bot
+from langbot.pkg.entity.persistence.base import Base
+from langbot.pkg.entity.persistence.metadata import WorkspaceMetadata
+from langbot.pkg.entity.persistence.pipeline import LegacyPipeline
+from langbot.pkg.entity.persistence.workspace import Workspace, WorkspaceExecutionState
+from langbot.pkg.persistence.mgr import PersistenceManager
+from langbot.pkg.pipeline.cntfilter.cntfilter import ContentFilterStage
+from langbot.pkg.pipeline.pipelinemgr import PipelineManager
+from langbot.pkg.utils.paths import get_resource_path
+from langbot.pkg.workspace.service import WorkspaceService
+from tests.integration.api.test_application_foundation import context
+
+
+@pytest.fixture
+async def write_foundation(tmp_path):
+    engine = create_async_engine(f'sqlite+aiosqlite:///{tmp_path / "writes.db"}')
+    try:
+        async with engine.begin() as connection:
+            await connection.run_sync(
+                lambda sync: Base.metadata.create_all(
+                    sync,
+                    tables=[
+                        model.__table__
+                        for model in (Workspace, WorkspaceExecutionState, WorkspaceMetadata, LegacyPipeline, Bot)
+                    ],
+                )
+            )
+            await connection.execute(
+                sa.insert(Workspace).values(
+                    uuid='workspace-a',
+                    instance_uuid='foundation-instance',
+                    name='Test',
+                    slug='test',
+                    source='cloud_projection',
+                )
+            )
+            await connection.execute(
+                sa.insert(WorkspaceExecutionState).values(
+                    workspace_uuid='workspace-a',
+                    instance_uuid='foundation-instance',
+                    active_generation=1,
+                    source='cloud',
+                )
+            )
+        ap = SimpleNamespace(instance_config=SimpleNamespace(data={}), logger=logging.getLogger('write-validation'))
+        ap.persistence_mgr = PersistenceManager(ap)
+        ap.persistence_mgr.db = SimpleNamespace(get_engine=lambda: engine)
+        ap.workspace_service = WorkspaceService(ap, instance_uuid='foundation-instance')
+        yield ApplicationAPI(ap), ap, engine
+    finally:
+        await engine.dispose()
+
+
+@pytest.fixture
+async def writes(write_foundation):
+    api, ap, engine = write_foundation
+    ap.pipeline_service = PipelineService(ap)
+    ap.bot_service = BotService(ap)
+    ap.pipeline_mgr = PipelineManager(ap)
+    ap.pipeline_mgr.stage_dict = {'ContentFilterStage': ContentFilterStage}
+    ap.sess_mgr = SimpleNamespace(session_list=[])
+    with open(get_resource_path('templates/default-pipeline-config.json')) as stream:
+        config = json.load(stream)
+    async with engine.begin() as connection:
+        await connection.execute(
+            sa.insert(LegacyPipeline).values(
+                uuid='pipeline-a',
+                workspace_uuid='workspace-a',
+                name='Original',
+                description='',
+                for_version='test',
+                stages=['ContentFilterStage'],
+                config=config,
+                extensions_preferences={},
+                is_default=True,
+            )
+        )
+    async with ap.persistence_mgr.tenant_scope('workspace-a'):
+        pipeline = await ap.pipeline_service.get_pipeline(context(), 'pipeline-a', include_secret=True)
+        await ap.pipeline_mgr.load_pipeline(context(), pipeline)
+    ap.user_service = SimpleNamespace(
+        get_authenticated_account=AsyncMock(return_value=SimpleNamespace(uuid='account-test', user='[email protected]'))
+    )
+    ap.apikey_service = SimpleNamespace(authenticate_api_key=AsyncMock(return_value=None))
+    ap.workspace_collaboration_service = SimpleNamespace(
+        resolve_account_workspace=AsyncMock(
+            return_value=SimpleNamespace(
+                workspace=SimpleNamespace(uuid='workspace-a'),
+                membership=SimpleNamespace(uuid='membership-test', role='owner', projection_revision=1),
+                execution=SimpleNamespace(instance_uuid='foundation-instance', placement_generation=1),
+            )
+        )
+    )
+    app = quart.Quart(__name__)
+    await PipelinesRouterGroup(ap, app).initialize()
+    await BotsRouterGroup(ap, app).initialize()
+    return api, ap, engine, app.test_client(), config
+
+
+async def stored_pipeline(engine):
+    async with engine.connect() as connection:
+        return dict((await connection.execute(sa.select(LegacyPipeline))).mappings().one())
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize('transport', ['http', 'application', 'service'])
+@pytest.mark.parametrize('invalid', ['partial', 'runtime', 'null'])
+async def test_invalid_pipeline_preserves_database_and_runtime(writes, transport, invalid):
+    api, ap, engine, client, config = writes
+    original = await stored_pipeline(engine)
+    runtime = ap.pipeline_mgr.pipelines[0]
+    if invalid == 'partial':
+        config = {'ai': config['ai']}
+    elif invalid == 'runtime':
+        config['safety'] = {}
+    else:
+        config = None
+    payload = {'name': 'Must not persist', 'config': config}
+    if transport == 'http':
+        response = await client.put(
+            '/api/v1/pipelines/pipeline-a', json=payload, headers={'Authorization': 'Bearer test-token'}
+        )
+        assert response.status_code == 400, await response.get_json()
+        print('pipeline.update', invalid, response.status_code, await response.get_json())
+    elif transport == 'application':
+        with pytest.raises(ApplicationAPIException) as failure:
+            await api.execute('pipeline.update', context(), {'pipeline_uuid': 'pipeline-a', 'pipeline_data': payload})
+        assert failure.value.error.status_code == 400
+    else:
+        async with ap.persistence_mgr.tenant_scope('workspace-a'):
+            with pytest.raises(ValueError):
+                await ap.pipeline_service.update_pipeline(context(), 'pipeline-a', payload)
+    assert await stored_pipeline(engine) == original
+    assert ap.pipeline_mgr.pipelines == [runtime]
+
+
+@pytest.mark.asyncio
+async def test_valid_pipeline_replaces_runtime_after_write(writes):
+    _, ap, engine, client, config = writes
+    previous = ap.pipeline_mgr.pipelines[0]
+    response = await client.put(
+        '/api/v1/pipelines/pipeline-a',
+        json={'name': 'Updated', 'config': config},
+        headers={'Authorization': 'Bearer test-token'},
+    )
+    assert response.status_code == 200, await response.get_json()
+    assert (await stored_pipeline(engine))['name'] == 'Updated'
+    assert ap.pipeline_mgr.pipelines[0] is not previous
+    assert ap.pipeline_mgr.pipelines[0].pipeline_entity.name == 'Updated'
+
+
+@pytest.mark.asyncio
+async def test_failed_pipeline_write_does_not_publish_candidate(writes, monkeypatch):
+    _, ap, engine, _, config = writes
+    original = await stored_pipeline(engine)
+    runtime = ap.pipeline_mgr.pipelines[0]
+    execute = ap.persistence_mgr.execute_async
+
+    async def fail_update(statement, *args, **kwargs):
+        if isinstance(statement, sa.sql.dml.Update):
+            raise RuntimeError('Database unavailable')
+        return await execute(statement, *args, **kwargs)
+
+    monkeypatch.setattr(ap.persistence_mgr, 'execute_async', fail_update)
+    async with ap.persistence_mgr.tenant_scope('workspace-a'):
+        with pytest.raises(RuntimeError, match='Database unavailable'):
+            await ap.pipeline_service.update_pipeline(context(), 'pipeline-a', {'config': config, 'name': 'Unsaved'})
+    assert await stored_pipeline(engine) == original
+    assert ap.pipeline_mgr.pipelines == [runtime]
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize('transport', ['http', 'application', 'service'])
+@pytest.mark.parametrize(
+    ('field', 'value'),
+    [
+        ('adapter_config', 'missing'),
+        ('adapter_config', None),
+        ('adapter_config', []),
+        ('adapter', 'missing'),
+        ('name', 'missing'),
+        ('adapter', ''),
+    ],
+)
+async def test_invalid_bot_has_no_database_side_effects(writes, transport, field, value):
+    api, ap, engine, client, _ = writes
+    pipeline = await stored_pipeline(engine)
+    payload = {'name': 'Bot', 'adapter': 'aiocqhttp', 'adapter_config': {}}
+    if value == 'missing':
+        payload.pop(field)
+    else:
+        payload[field] = value
+    original = copy.deepcopy(payload)
+    if transport == 'http':
+        response = await client.post(
+            '/api/v1/platform/bots', json=payload, headers={'Authorization': 'Bearer test-token'}
+        )
+        assert response.status_code == 400, await response.get_json()
+        assert field in (await response.get_json())['msg']
+        print('bot.create', field, response.status_code, await response.get_json())
+    elif transport == 'application':
+        with pytest.raises(ApplicationAPIException) as failure:
+            await api.execute('bot.create', context(), {'bot_data': payload})
+        assert failure.value.error.status_code == 400
+        assert field in failure.value.error.message
+    else:
+        async with ap.persistence_mgr.tenant_scope('workspace-a'):
+            with pytest.raises(ValueError, match=field):
+                await ap.bot_service.create_bot(context(), payload)
+    async with engine.connect() as connection:
+        assert (await connection.execute(sa.select(sa.func.count()).select_from(Bot))).scalar_one() == 0
+    assert await stored_pipeline(engine) == pipeline
+    assert payload == original
+
+
+@pytest.mark.asyncio
+async def test_bot_create_schema_lists_required_fields(writes):
+    api, _, _, _, _ = writes
+    operation = next(item for item in api.operation_catalog() if item['id'] == 'bot.create')
+    schema = operation['input_schema']['properties']['bot_data']
+    assert schema['required'] == ['name', 'adapter', 'adapter_config']
+    assert schema['properties']['adapter_config']['type'] == 'object'
+
+
+@pytest.mark.asyncio
+async def test_valid_bot_binds_existing_pipeline(writes):
+    _, _, engine, client, _ = writes
+    response = await client.post(
+        '/api/v1/platform/bots',
+        json={
+            'name': 'Bound bot',
+            'adapter': 'aiocqhttp',
+            'adapter_config': {},
+            'enable': False,
+        },
+        headers={'Authorization': 'Bearer test-token'},
+    )
+    assert response.status_code == 200, await response.get_json()
+    async with engine.connect() as connection:
+        bot = (await connection.execute(sa.select(Bot))).mappings().one()
+    assert bot['adapter_config'] == {}
+    assert all(binding['target_uuid'] == 'pipeline-a' for binding in bot['event_bindings'])
+    print('bot.create valid', response.status_code, await response.get_json(), 'bindings:', bot['event_bindings'])
+
+
+@pytest.mark.asyncio
+async def test_live_http_rejects_invalid_writes_without_dirty_rows(writes):
+    _, ap, engine, client, _ = writes
+    original = await stored_pipeline(engine)
+    runtime = ap.pipeline_mgr.pipelines[0]
+    listener = socket.socket()
+    listener.bind(('127.0.0.1', 0))
+    listener.listen()
+    port = listener.getsockname()[1]
+    config = Config()
+    config.bind = [f'fd://{listener.detach()}']
+    shutdown = asyncio.Event()
+    server = asyncio.create_task(serve(client.app, config, shutdown_trigger=shutdown.wait))
+    try:
+        async with aiohttp.ClientSession(timeout=aiohttp.ClientTimeout(total=15)) as session:
+            for method, path, payload in (
+                ('PUT', '/api/v1/pipelines/pipeline-a', {'name': 'Unsaved', 'config': {'ai': {}}}),
+                ('POST', '/api/v1/platform/bots', {'name': 'Invalid bot', 'adapter': 'aiocqhttp'}),
+            ):
+                async with session.request(
+                    method,
+                    f'http://127.0.0.1:{port}{path}',
+                    json=payload,
+                    headers={'Authorization': 'Bearer test-token'},
+                ) as response:
+                    body = await response.json()
+                    print(
+                        json.dumps(
+                            {
+                                'method': method,
+                                'path': path,
+                                'request': payload,
+                                'status': response.status,
+                                'response': body,
+                            }
+                        )
+                    )
+                    assert response.status == 400
+        assert await stored_pipeline(engine) == original
+        async with engine.connect() as connection:
+            assert (await connection.execute(sa.select(sa.func.count()).select_from(Bot))).scalar_one() == 0
+        assert ap.pipeline_mgr.pipelines == [runtime]
+        print('Independent SQLite readback: pipeline unchanged; bots=0; original runtime retained')
+    finally:
+        shutdown.set()
+        await server

tests/unit_tests/api/service/test_bot_service.py

@@ -369,7 +369,7 @@ class TestBotServiceCreateBot:
 
         # Execute & Verify
         with pytest.raises(ValueError, match='Maximum number of bots'):
-            await service.create_bot(WORKSPACE_UUID, {'name': 'New Bot'})
+            await service.create_bot(WORKSPACE_UUID, {'name': 'New Bot', 'adapter': 'aiocqhttp', 'adapter_config': {}})
         assert all(not call.args[0].is_insert for call in ap.persistence_mgr.execute_async.await_args_list)
         ap.platform_mgr.load_bot.assert_not_awaited()
 

tests/unit_tests/api/service/test_pipeline_service.py

@@ -411,6 +411,8 @@ class TestPipelineServiceUpdatePipeline:
         ap.pipeline_mgr = SimpleNamespace()
         ap.pipeline_mgr.remove_pipeline = AsyncMock()
         ap.pipeline_mgr.load_pipeline = AsyncMock()
+        ap.pipeline_mgr.prepare_pipeline = AsyncMock()
+        ap.pipeline_mgr.publish_pipeline = Mock()
         ap.sess_mgr = SimpleNamespace()
         ap.sess_mgr.session_list = []
         ap.bot_service = None  # No bot_service when not updating name
@@ -444,6 +446,8 @@ class TestPipelineServiceUpdatePipeline:
         ap.pipeline_mgr = SimpleNamespace()
         ap.pipeline_mgr.remove_pipeline = AsyncMock()
         ap.pipeline_mgr.load_pipeline = AsyncMock()
+        ap.pipeline_mgr.prepare_pipeline = AsyncMock()
+        ap.pipeline_mgr.publish_pipeline = Mock()
         ap.sess_mgr = SimpleNamespace()
         ap.sess_mgr.session_list = []
         ap.bot_service = SimpleNamespace()
@@ -497,6 +501,8 @@ class TestPipelineServiceUpdatePipeline:
         ap.pipeline_mgr = SimpleNamespace()
         ap.pipeline_mgr.remove_pipeline = AsyncMock()
         ap.pipeline_mgr.load_pipeline = AsyncMock()
+        ap.pipeline_mgr.prepare_pipeline = AsyncMock()
+        ap.pipeline_mgr.publish_pipeline = Mock()
         ap.sess_mgr = SimpleNamespace()
 
         # Mock session with conversation using this pipeline
@@ -965,7 +971,7 @@ class TestPipelineSecretRoundtrip:
         write_result = Mock(rowcount=1)
         ap = SimpleNamespace(
             persistence_mgr=SimpleNamespace(execute_async=AsyncMock(return_value=write_result)),
-            pipeline_mgr=SimpleNamespace(remove_pipeline=AsyncMock(), load_pipeline=AsyncMock()),
+            pipeline_mgr=SimpleNamespace(prepare_pipeline=AsyncMock(), publish_pipeline=Mock()),
             sess_mgr=SimpleNamespace(session_list=[]),
         )
         service = PipelineService(ap)
@@ -976,13 +982,16 @@ class TestPipelineSecretRoundtrip:
             'pipeline-secret',
             {
                 'config': {
+                    'trigger': {},
+                    'safety': {},
+                    'output': {},
                     'ai': {
                         'n8n': {
                             'webhook-url': '***',
                             'headers': {'Authorization': '***'},
                             'timeout': 60,
                         }
-                    }
+                    },
                 }
             },
         )
@@ -990,6 +999,9 @@ class TestPipelineSecretRoundtrip:
         statement = ap.persistence_mgr.execute_async.await_args.args[0]
         stored_config = next(value.value for column, value in statement._values.items() if column.key == 'config')
         assert stored_config == {
+            'trigger': {},
+            'safety': {},
+            'output': {},
             'ai': {
                 'n8n': {
                     'webhook-url': 'https://hook.invalid/bearer-secret',

tests/unit_tests/pipeline/test_pipeline_service.py

@@ -11,14 +11,14 @@ WORKSPACE_UUID = 'workspace-a'
 @pytest.mark.asyncio
 async def test_update_pipeline_filters_protected_fields_without_mutating_input(mock_app):
     service = PipelineService(mock_app)
-    loaded_pipeline = Mock()
+    loaded_pipeline = {'uuid': 'pipeline-uuid'}
     service.get_pipeline = AsyncMock(return_value=loaded_pipeline)
 
     bot = Mock(uuid='bot-uuid')
     bot_result = Mock(all=Mock(return_value=[bot]))
     mock_app.persistence_mgr.execute_async = AsyncMock(side_effect=[None, bot_result])
     mock_app.bot_service = Mock(update_bot=AsyncMock())
-    mock_app.pipeline_mgr = Mock(remove_pipeline=AsyncMock(), load_pipeline=AsyncMock())
+    mock_app.pipeline_mgr = Mock(prepare_pipeline=AsyncMock(), publish_pipeline=Mock())
     mock_app.sess_mgr.session_list = []
 
     pipeline_data = {
@@ -39,5 +39,7 @@ async def test_update_pipeline_filters_protected_fields_without_mutating_input(m
     assert updated_fields == {'name'}
 
     mock_app.bot_service.update_bot.assert_not_awaited()
-    mock_app.pipeline_mgr.remove_pipeline.assert_awaited_once_with('workspace-a', 'pipeline-uuid')
-    mock_app.pipeline_mgr.load_pipeline.assert_awaited_once_with('workspace-a', loaded_pipeline)
+    mock_app.pipeline_mgr.prepare_pipeline.assert_awaited_once_with(
+        'workspace-a', {'uuid': 'pipeline-uuid', 'name': 'Updated pipeline'}
+    )
+    mock_app.pipeline_mgr.publish_pipeline.assert_called_once_with(mock_app.pipeline_mgr.prepare_pipeline.return_value)