test(api): verify orchestration writes and scoped state transitions

tests/unit_tests/api/service/test_bot_service.py

@@ -11,12 +11,15 @@ from __future__ import annotations
 import pytest
 from unittest.mock import AsyncMock, MagicMock, Mock, patch
 from types import SimpleNamespace
+import hashlib
+import hmac
 import json
 import sqlalchemy
 import uuid
 
 from langbot.pkg.api.http.service.bot import BotService
 from langbot.pkg.entity.persistence.bot import Bot
+from langbot.pkg.workspace.errors import WorkspaceNotFoundError
 
 
 pytestmark = pytest.mark.asyncio
@@ -182,6 +185,11 @@ class TestBotServiceGetBot:
         assert result is not None
         assert result['uuid'] == 'test-uuid'
         assert result['name'] == 'Found Bot'
+        statement = ap.persistence_mgr.execute_async.await_args.args[0]
+        assert statement.whereclause.compare(
+            sqlalchemy.and_(Bot.uuid == 'test-uuid', Bot.workspace_uuid == WORKSPACE_UUID)
+        )
+        ap.persistence_mgr.serialize_model.assert_called_once_with(Bot, bot, ['adapter_config'])
 
     async def test_get_bot_by_uuid_not_found(self):
         """Returns None when bot not found."""
@@ -361,6 +369,8 @@ class TestBotServiceCreateBot:
         # Execute & Verify
         with pytest.raises(ValueError, match='Maximum number of bots'):
             await service.create_bot(WORKSPACE_UUID, {'name': 'New Bot'})
+        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()
 
     async def test_create_bot_no_limit(self):
         """Creates bot without limit check when max_bots=-1."""
@@ -408,7 +418,22 @@ class TestBotServiceCreateBot:
 
         # Verify
         assert bot_uuid is not None
-        assert len(bot_uuid) == 36  # UUID format
+        assert uuid.UUID(bot_uuid).version == 4
+        statement = ap.persistence_mgr.execute_async.await_args_list[0].args[0]
+        assert statement.is_insert
+        assert statement.table.name == Bot.__tablename__
+        values = statement.compile().params
+        expected = {
+            'uuid': bot_uuid,
+            'workspace_uuid': WORKSPACE_UUID,
+            'name': 'New Bot',
+            'adapter': 'telegram',
+            'adapter_config': {},
+            'event_bindings': [],
+            'plugin_processors': [],
+        }
+        assert {key: values.get(key) for key in expected} == expected
+        ap.platform_mgr.load_bot.assert_not_awaited()
 
     async def test_create_bot_scopes_insert_without_legacy_pipeline_fields(self):
         """Creates a Workspace-owned Bot without restoring removed pipeline columns."""
@@ -647,26 +672,28 @@ class TestBotServiceDeleteBot:
         await service.delete_bot(WORKSPACE_UUID, 'test-uuid')
 
         # Verify
-        ap.platform_mgr.remove_bot.assert_called_once_with(WORKSPACE_UUID, 'test-uuid')
-        ap.persistence_mgr.execute_async.assert_called_once()
+        ap.platform_mgr.remove_bot.assert_awaited_once_with(WORKSPACE_UUID, 'test-uuid')
+        ap.persistence_mgr.execute_async.assert_awaited_once()
+        statement = ap.persistence_mgr.execute_async.await_args.args[0]
+        assert statement.is_delete
+        assert statement.table.name == Bot.__tablename__
+        assert statement.whereclause.compare(
+            sqlalchemy.and_(Bot.uuid == 'test-uuid', Bot.workspace_uuid == WORKSPACE_UUID)
+        )
 
     async def test_delete_bot_nonexistent_uuid(self):
-        """Delete operation completes even for nonexistent UUID."""
+        """A missing persisted Bot is rejected before removing its runtime."""
         # Setup
         ap = SimpleNamespace()
         ap.persistence_mgr = SimpleNamespace()
-        ap.persistence_mgr.execute_async = AsyncMock()
+        ap.persistence_mgr.execute_async = AsyncMock(return_value=SimpleNamespace(rowcount=0))
         ap.platform_mgr = SimpleNamespace()
         ap.platform_mgr.remove_bot = AsyncMock()
 
         service = BotService(ap)
-        service.get_bot = AsyncMock(return_value={'uuid': 'bot-uuid'})
-
-        # Execute - should not raise
-        await service.delete_bot(WORKSPACE_UUID, 'nonexistent-uuid')
-
-        # Verify - both called regardless
-        ap.platform_mgr.remove_bot.assert_called_once()
+        with pytest.raises(WorkspaceNotFoundError, match='Bot not found'):
+            await service.delete_bot(WORKSPACE_UUID, 'nonexistent-uuid')
+        ap.platform_mgr.remove_bot.assert_not_awaited()
 
 
 class TestBotServiceListEventLogs:
@@ -811,7 +838,9 @@ class TestBotServiceHttpBotInboundTest:
         assert payload['message'] == [{'type': 'Plain', 'text': 'hello'}]
         headers = request.kwargs['headers']
         assert headers['X-LB-Timestamp']
-        assert headers['X-LB-Signature'].startswith('sha256=')
+        signing_input = headers['X-LB-Timestamp'].encode() + b'.' + request.kwargs['data']
+        digest = hmac.new(b'test-secret', signing_input, hashlib.sha256).hexdigest()
+        assert headers['X-LB-Signature'] == f'sha256={digest}'
 
     async def test_rejects_non_http_bot(self):
         service = BotService(SimpleNamespace())

tests/unit_tests/api/service/test_pipeline_migration.py

@@ -367,6 +367,9 @@ async def test_ordinary_update_requires_manual_migration_but_allows_metadata(env
         await service.update_pipeline(context(), 'one', {'name': 'renamed'})
     configs, backups = await rows(env)
     assert configs['one'] == SOURCE and not backups
+    async with env.engine.connect() as conn:
+        name = (await conn.execute(sa.select(LegacyPipeline.name).where(LegacyPipeline.uuid == 'one'))).scalar_one()
+    assert name == 'renamed'
 
 
 @pytest.mark.asyncio
@@ -777,7 +780,8 @@ async def test_all_mode_installs_missing_plugin_once_then_migrates_both(env):
         await conn.execute(sa.delete(PluginSetting))
 
     async def install(source, info, task_context):
-        assert info['plugin_version'] == '1.0'
+        assert source == env.m.PluginInstallSource.MARKETPLACE
+        assert info == {'plugin_author': 'langbot-team', 'plugin_name': 'TestAgent', 'plugin_version': '1.0'}
         async with env.engine.begin() as conn:
             await conn.execute(
                 sa.insert(PluginSetting).values(
@@ -848,7 +852,14 @@ async def test_all_mode_has_no_fifty_pipeline_limit(env):
         )
     response, metadata = await execute_all(env, install_plugins=False)
     assert len(response['pipeline_uuids']) == 53
+    expected = {'one', 'two', *(f'extra-{i}' for i in range(51))}
+    assert {r['pipeline_uuid'] for r in metadata['results']} == expected
+    assert len(metadata['results']) == 53
     assert all(r['state'] == 'migrated' for r in metadata['results'])
+    configs, backups = await rows(env)
+    assert len(backups) == 53
+    assert all(configs[pipeline_uuid]['ai']['runner']['id'] == RID for pipeline_uuid in expected)
+    assert configs['foreign'] == SOURCE
 
 
 @pytest.mark.asyncio

tests/unit_tests/api/service/test_pipeline_service.py

@@ -17,6 +17,7 @@ import pytest
 from unittest.mock import AsyncMock, Mock, patch, mock_open
 from types import SimpleNamespace
 import uuid
+import sqlalchemy
 import json
 
 from langbot.pkg.api.http.service.pipeline import PipelineService, default_stage_order
@@ -145,8 +146,12 @@ class TestPipelineServiceGetPipelines:
         )
 
         # Verify
-        assert len(result) == 2
-        ap.persistence_mgr.execute_async.assert_called_once()
+        assert result == [{'uuid': 'uuid-1', 'name': 'Pipeline 1'}, {'uuid': 'uuid-2', 'name': 'Pipeline 2'}]
+        ap.persistence_mgr.execute_async.assert_awaited_once()
+        statement = ap.persistence_mgr.execute_async.await_args.args[0]
+        assert len(statement._order_by_clauses) == 1
+        assert next(iter(statement._order_by_clauses)).compare(LegacyPipeline.created_at.desc())
+        assert statement.whereclause.compare(LegacyPipeline.workspace_uuid == WORKSPACE_UUID)
 
     async def test_get_pipelines_sort_by_updated_at_asc(self):
         """Returns pipelines sorted by updated_at ascending."""
@@ -164,7 +169,11 @@ class TestPipelineServiceGetPipelines:
         await service.get_pipelines(WORKSPACE_UUID, sort_by='updated_at', sort_order='ASC')
 
         # Verify - execute was called with sort parameters
-        ap.persistence_mgr.execute_async.assert_called_once()
+        ap.persistence_mgr.execute_async.assert_awaited_once()
+        statement = ap.persistence_mgr.execute_async.await_args.args[0]
+        assert len(statement._order_by_clauses) == 1
+        assert next(iter(statement._order_by_clauses)).compare(LegacyPipeline.updated_at.asc())
+        assert statement.whereclause.compare(LegacyPipeline.workspace_uuid == WORKSPACE_UUID)
 
 
 class TestPipelineServiceGetPipeline:
@@ -271,7 +280,19 @@ class TestPipelineServiceCreatePipeline:
 
         # Verify
         assert bot_uuid is not None
-        assert len(bot_uuid) == 36  # UUID format
+        assert uuid.UUID(bot_uuid).version == 4
+        statement = ap.persistence_mgr.execute_async.await_args.args[0]
+        assert statement.is_insert
+        assert statement.table.name == LegacyPipeline.__tablename__
+        values = statement.compile().params
+        assert values['uuid'] == bot_uuid
+        assert values['workspace_uuid'] == WORKSPACE_UUID
+        assert values['name'] == 'New Pipeline'
+        assert values['config'] == default_config
+        assert values['for_version'] == '1.0.0'
+        assert values['is_default'] is False
+        service.get_pipeline.assert_awaited_once_with(WORKSPACE_UUID, bot_uuid, include_secret=True)
+        ap.pipeline_mgr.load_pipeline.assert_awaited_once_with(WORKSPACE_UUID, service.get_pipeline.return_value)
 
     async def test_create_pipeline_as_default(self):
         """Creates pipeline with is_default=True."""
@@ -305,7 +326,10 @@ class TestPipelineServiceCreatePipeline:
                 await service.create_pipeline(WORKSPACE_UUID, {'name': 'Default Pipeline'}, default=True)
 
         # Verify - execute was called
-        ap.persistence_mgr.execute_async.assert_called()
+        ap.persistence_mgr.execute_async.assert_awaited_once()
+        statement = ap.persistence_mgr.execute_async.await_args.args[0]
+        assert statement.is_insert
+        assert statement.compile().params['is_default'] is True
 
     async def test_create_pipeline_sets_default_extensions_preferences(self):
         """Sets default extensions_preferences when not provided."""
@@ -515,21 +539,19 @@ class TestPipelineServiceDeletePipeline:
         ap.persistence_mgr.execute_async.assert_called_once()
 
     async def test_delete_pipeline_nonexistent_uuid(self):
-        """Delete operation completes even for nonexistent UUID."""
+        """Missing persisted Pipelines do not remove runtime instances."""
         # Setup
         ap = SimpleNamespace()
         ap.persistence_mgr = SimpleNamespace()
-        ap.persistence_mgr.execute_async = AsyncMock()
+        ap.persistence_mgr.execute_async = AsyncMock(return_value=SimpleNamespace(rowcount=0))
         ap.pipeline_mgr = SimpleNamespace()
         ap.pipeline_mgr.remove_pipeline = AsyncMock()
 
         service = PipelineService(ap)
 
-        # Execute - should not raise
-        await service.delete_pipeline(WORKSPACE_UUID, 'nonexistent-uuid')
-
-        # Verify
-        ap.pipeline_mgr.remove_pipeline.assert_called_once()
+        with pytest.raises(WorkspaceNotFoundError, match='Pipeline not found'):
+            await service.delete_pipeline(WORKSPACE_UUID, 'nonexistent-uuid')
+        ap.pipeline_mgr.remove_pipeline.assert_not_awaited()
 
 
 class TestPipelineServiceCopyPipeline:
@@ -627,7 +649,23 @@ class TestPipelineServiceCopyPipeline:
 
         # Verify
         assert new_uuid is not None
-        assert len(new_uuid) == 36  # UUID format
+        assert uuid.UUID(new_uuid).version == 4
+        insert = ap.persistence_mgr.execute_async.await_args_list[1].args[0]
+        assert insert.is_insert
+        values = insert.compile().params
+        assert values['uuid'] == new_uuid
+        assert values['workspace_uuid'] == WORKSPACE_UUID
+        assert values['name'] == 'Original Pipeline (Copy)'
+        assert values['description'] == 'Original description'
+        assert values['stages'] == ['Stage1', 'Stage2']
+        assert values['config'] == {'key': 'value'}
+        assert values['extensions_preferences']['enable_all_plugins'] is False
+        assert values['extensions_preferences']['plugins'] == []
+        query = ap.persistence_mgr.execute_async.await_args_list[0].args[0]
+        assert query.whereclause.compare(
+            sqlalchemy.and_(LegacyPipeline.uuid == 'original-uuid', LegacyPipeline.workspace_uuid == WORKSPACE_UUID)
+        )
+        service.get_pipeline.assert_awaited_once_with(WORKSPACE_UUID, new_uuid, include_secret=True)
 
     async def test_copy_pipeline_is_not_default(self):
         """Copy is never set as default."""
@@ -659,7 +697,20 @@ class TestPipelineServiceCopyPipeline:
         await service.copy_pipeline(WORKSPACE_UUID, 'original-uuid')
 
         # Verify - pipeline_mgr.load_pipeline called (copy created)
-        ap.pipeline_mgr.load_pipeline.assert_called_once()
+        ap.pipeline_mgr.load_pipeline.assert_awaited_once_with(WORKSPACE_UUID, service.get_pipeline.return_value)
+        insert = ap.persistence_mgr.execute_async.await_args_list[1].args[0]
+        assert insert.is_insert
+        assert insert.compile().params['is_default'] is False
+
+
+def _persisted_extensions(ap):
+    updates = [call.args[0] for call in ap.persistence_mgr.execute_async.await_args_list if call.args[0].is_update]
+    assert len(updates) == 1
+    statement = updates[0]
+    assert statement.whereclause.compare(
+        sqlalchemy.and_(LegacyPipeline.uuid == 'test-uuid', LegacyPipeline.workspace_uuid == WORKSPACE_UUID)
+    )
+    return statement.compile().params['extensions_preferences']
 
 
 class TestPipelineServiceUpdatePipelineExtensions:
@@ -731,7 +782,9 @@ class TestPipelineServiceUpdatePipelineExtensions:
         )
 
         # Verify
-        ap.persistence_mgr.execute_async.assert_called()
+        persisted = _persisted_extensions(ap)
+        assert persisted['enable_all_plugins'] is False
+        assert persisted['plugins'] == [{'author': 'test', 'name': 'plugin-1'}]
 
     async def test_update_extensions_sets_mcp_servers(self):
         """Updates MCP servers in extensions_preferences."""
@@ -782,7 +835,9 @@ class TestPipelineServiceUpdatePipelineExtensions:
         )
 
         # Verify
-        ap.persistence_mgr.execute_async.assert_called()
+        persisted = _persisted_extensions(ap)
+        assert persisted['enable_all_mcp_servers'] is False
+        assert persisted['mcp_servers'] == ['mcp-server-1']
 
     async def test_update_extensions_none_mcp_servers_keeps_existing(self):
         """Does not modify mcp_servers when bound_mcp_servers is None."""
@@ -825,7 +880,7 @@ class TestPipelineServiceUpdatePipelineExtensions:
         await service.update_pipeline_extensions(WORKSPACE_UUID, 'test-uuid', bound_plugins=[])
 
         # Verify - persistence was called
-        ap.persistence_mgr.execute_async.assert_called()
+        assert _persisted_extensions(ap)['mcp_servers'] == ['existing-server']
 
     async def test_update_extensions_preserves_mcp_resource_agent_read_when_omitted(self):
         """Does not reset mcp_resource_agent_read_enabled when omitted by older clients."""
@@ -863,10 +918,9 @@ class TestPipelineServiceUpdatePipelineExtensions:
 
         await service.update_pipeline_extensions(WORKSPACE_UUID, 'test-uuid', bound_plugins=[])
 
-        assert original_pipeline.extensions_preferences['mcp_resource_agent_read_enabled'] is False
-        assert original_pipeline.extensions_preferences['mcp_resources'] == [
-            {'server_uuid': 'srv-1', 'uri': 'file:///README.md'}
-        ]
+        persisted = _persisted_extensions(ap)
+        assert persisted['mcp_resource_agent_read_enabled'] is False
+        assert persisted['mcp_resources'] == [{'server_uuid': 'srv-1', 'uri': 'file:///README.md'}]
 
 
 class TestPipelineSecretRoundtrip:

tests/unit_tests/api/service/test_session_context_reset.py

@@ -86,11 +86,19 @@ async def test_reset_preserves_logs_and_isolates_context(context_service):
     async with engine.connect() as conn:
         assert (await conn.execute(sa.select(sa.func.count()).select_from(Transcript))).scalar() == 2
         assert (await conn.execute(sa.select(sa.func.count()).select_from(MonitoringSession))).scalar() == 1
-        assert (
+        remaining = (
             await conn.execute(
-                sa.select(sa.func.count()).select_from(RunnerState).where(RunnerState.runner_id == 'runner')
+                sa.select(
+                    RunnerState.workspace_id, RunnerState.bot_id, RunnerState.conversation_id, RunnerState.scope
+                ).where(RunnerState.runner_id == 'runner')
             )
-        ).scalar() == 4
+        ).all()
+        assert set(remaining) == {
+            ('w', 'b2', 'person_u', 'conversation'),
+            ('w2', 'b', 'person_u', 'conversation'),
+            ('w', 'b', 'person_other', 'conversation'),
+            ('w', 'b', 'person_u', 'actor'),
+        }
     page = await store.page_transcript('person_u', bot_id='b', workspace_id='w')
     assert not page[0]
     assert (await store.page_transcript('person_u', bot_id='b2', workspace_id='w'))[0]
@@ -143,6 +151,6 @@ async def test_reset_starts_new_external_runner_conversation(context_service):
     ] == 'remote-old'
     await service.reset_session_context('w', 'b', 'person_u')
     snapshot = await store.build_snapshot_from_event(event, binding, descriptor)
-    assert snapshot['conversation']['external.conversation_id'] == await get_reset_generation(
-        engine, 'w', 'b', 'person_u'
-    )
+    generation = await get_reset_generation(engine, 'w', 'b', 'person_u')
+    assert isinstance(generation, str) and generation and generation not in {'remote-old', 'person_u'}
+    assert snapshot['conversation']['external.conversation_id'] == generation

tests/unit_tests/api/test_adapter_session_scoping.py

@@ -190,10 +190,17 @@ async def test_session_capacity_evicts_oldest_session_in_same_workspace():
         _bind_session_scope(session, owner_context)
         sessions[f'session-{index}'] = session
 
+    foreign_task = SimpleNamespace(done=Mock(return_value=False), cancel=Mock())
+    foreign = {'created_at': -1.0, 'task': foreign_task}
+    _bind_session_scope(foreign, _request_context(workspace_uuid='workspace-b'))
+    sessions['foreign'] = foreign
+
     _make_room_for_session(sessions, owner_context)
 
     assert 'session-0' not in sessions
-    assert len(sessions) == 9
+    assert len(sessions) == 10
+    assert sessions['foreign'] is foreign
+    foreign_task.cancel.assert_not_called()
     tasks[0].cancel.assert_called_once_with()