test(api): verify backfill, section updates and asynchronous activation

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

tests/integration/api/test_pipeline_bot_write_validation.py

@@ -125,13 +125,13 @@ async def stored_pipeline(engine):
 
 @pytest.mark.asyncio
 @pytest.mark.parametrize('transport', ['http', 'application', 'service'])
-@pytest.mark.parametrize('invalid', ['partial', 'runtime', 'null'])
+@pytest.mark.parametrize('invalid', ['non_object_section', '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']}
+    if invalid == 'non_object_section':
+        config = {'ai': []}
     elif invalid == 'runtime':
         config['safety'] = {}
     else:
@@ -155,6 +155,26 @@ async def test_invalid_pipeline_preserves_database_and_runtime(writes, transport
     assert ap.pipeline_mgr.pipelines == [runtime]
 
 
+@pytest.mark.asyncio
+@pytest.mark.parametrize('transport', ['http', 'application', 'service'])
+async def test_partial_pipeline_update_preserves_omitted_sections(writes, transport):
+    api, ap, engine, client, config = writes
+    payload = {'config': {'ai': {}}}
+    if transport == 'http':
+        response = await client.put(
+            '/api/v1/pipelines/pipeline-a', json=payload, headers={'Authorization': 'Bearer test-token'}
+        )
+        assert response.status_code == 200, await response.get_json()
+    elif transport == 'application':
+        await api.execute('pipeline.update', context(), {'pipeline_uuid': 'pipeline-a', 'pipeline_data': payload})
+    else:
+        async with ap.persistence_mgr.tenant_scope('workspace-a'):
+            await ap.pipeline_service.update_pipeline(context(), 'pipeline-a', payload)
+    stored = await stored_pipeline(engine)
+    assert stored['config'] == {**config, 'ai': {}}
+    assert ap.pipeline_mgr.pipelines[0].pipeline_entity.config == stored['config']
+
+
 @pytest.mark.asyncio
 async def test_valid_pipeline_replaces_runtime_after_write(writes):
     _, ap, engine, client, config = writes
@@ -280,7 +300,7 @@ async def test_live_http_rejects_invalid_writes_without_dirty_rows(writes):
     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': {}}}),
+                ('PUT', '/api/v1/pipelines/pipeline-a', {'name': 'Unsaved', 'config': {'ai': []}}),
                 ('POST', '/api/v1/platform/bots', {'name': 'Invalid bot', 'adapter': 'aiocqhttp'}),
             ):
                 async with session.request(

tests/unit_tests/persistence/test_plugin_permissions_migration.py

@@ -1,6 +1,7 @@
 """The permission migration supports both upgraded and fresh schemas."""
 
 import importlib
+import json
 
 from alembic.migration import MigrationContext
 from alembic.operations import Operations
@@ -24,7 +25,12 @@ def test_plugin_permission_columns_preserve_existing_installs(fresh_schema):
             migration.upgrade()
         existing = {column['name'] for column in sa.inspect(connection).get_columns('plugin_settings')}
         assert set(names) <= existing
-        assert (
-            connection.execute(sa.text('SELECT permission_grants FROM plugin_settings WHERE id = 1')).scalar() is None
-        )
+        row = connection.execute(
+            sa.text(
+                'SELECT permission_declaration, permission_grants, permission_snapshot FROM plugin_settings WHERE id = 1'
+            )
+        ).one()
+        expected = {'workspace_operations': [], 'capabilities': migration.LEGACY_CAPABILITIES}
+        assert row.permission_declaration is None
+        assert json.loads(row.permission_grants) == json.loads(row.permission_snapshot) == expected
     engine.dispose()

tools/entrypoints/regression_ci.py

@@ -358,7 +358,13 @@ def run(args):
                 time.sleep(0.2)
             raise TimeoutError(f'Plugin installation/confirmation timed out: {entry["id"]}')
 
-        install_plugins(suite.call, root / 'plugins', packages, install=install_confirmed_plugin)
+        install_plugins(
+            suite.call,
+            root / 'plugins',
+            packages,
+            install=install_confirmed_plugin,
+            readiness_timeout=suite.timeout(180),
+        )
         restart_result = None
         restart_elapsed = 0
         if args.restart_services:

tools/regression/regression_plugins.py

@@ -13,6 +13,7 @@ import subprocess
 import sys
 import tempfile
 import threading
+import time
 import urllib.error
 import urllib.request
 import zipfile
@@ -236,7 +237,7 @@ def package_marketplace(directory, manifest):
         thread.join()
 
 
-def install_plugins(call, directory, manifest, *, install=None):
+def install_plugins(call, directory, manifest, *, install=None, readiness_timeout=0):
     """Install recorded packages and require exact, initialized CLI readback."""
     directory = Path(directory)
     for entry in manifest['plugins']:
@@ -250,17 +251,23 @@ def install_plugins(call, directory, manifest, *, install=None):
                 install(entry)
         except RuntimeError as exc:
             raise RuntimeError(f'Plugin install failed: {entry["id"]}@{entry["version"]}: {exc}') from exc
-        response = call(['plugin', 'list'])
-        plugins = response.get('plugins') if isinstance(response, dict) else None
-        if not isinstance(plugins, list) or any(not isinstance(item, dict) for item in plugins):
-            raise RuntimeError('Plugin readback did not return a plugin list')
-        matches = [item for item in plugins if f'{item.get("author")}/{item.get("name")}' == entry['id']]
+        deadline = time.monotonic() + readiness_timeout
+        while True:
+            response = call(['plugin', 'list'])
+            plugins = response.get('plugins') if isinstance(response, dict) else None
+            if not isinstance(plugins, list) or any(not isinstance(item, dict) for item in plugins):
+                raise RuntimeError('Plugin readback did not return a plugin list')
+            matches = [item for item in plugins if f'{item.get("author")}/{item.get("name")}' == entry['id']]
+            ready = (
+                len(matches) == 1
+                and matches[0].get('version') == entry['version']
+                and matches[0].get('enabled') is True
+                and matches[0].get('status') == 'initialized'
+            )
+            if ready or time.monotonic() >= deadline:
+                break
+            time.sleep(0.2)
         entry['installed'] = [{key: item.get(key) for key in ('version', 'enabled', 'status')} for item in matches]
         write_manifest(directory, manifest)
-        if (
-            len(matches) != 1
-            or matches[0].get('version') != entry['version']
-            or matches[0].get('enabled') is not True
-            or matches[0].get('status') != 'initialized'
-        ):
+        if not ready:
             raise RuntimeError(f'Plugin is not ready at the selected version: {entry["id"]}: {entry["installed"]}')

tools/tests/test_regression_plugins.py

@@ -265,6 +265,49 @@ class PublishedPluginsTests(unittest.TestCase):
             plugins.install_plugins(call, directory, manifest)
         self.assertEqual(len(calls), 2)
 
+    def test_confirmed_install_waits_for_worker_initialization(self):
+        directory = self.root / 'latest'
+        manifest = plugins.prepare_published_plugins(directory)
+        manifest['plugins'] = manifest['plugins'][:1]
+        installed = {'author': 'langbot-team', 'name': 'LocalAgent', 'version': '1.2.3', 'enabled': True}
+        responses = iter(
+            [
+                {'plugins': []},
+                {'plugins': [{**installed, 'status': 'initializing'}]},
+                {'plugins': [{**installed, 'status': 'initialized'}]},
+            ]
+        )
+        with patch.object(plugins.time, 'sleep') as sleep:
+            plugins.install_plugins(
+                lambda args: next(responses), directory, manifest, install=lambda entry: None, readiness_timeout=10
+            )
+        self.assertEqual(sleep.call_count, 2)
+        self.assertEqual(manifest['plugins'][0]['installed'][0]['status'], 'initialized')
+
+    def test_confirmed_install_pending_permissions_times_out(self):
+        directory = self.root / 'latest'
+        manifest = plugins.prepare_published_plugins(directory)
+        pending = {
+            'author': 'langbot-team',
+            'name': 'LocalAgent',
+            'version': '1.2.3',
+            'enabled': False,
+            'status': 'pending_confirmation',
+        }
+        with (
+            patch.object(plugins.time, 'monotonic', side_effect=[0, 0, 11]),
+            patch.object(plugins.time, 'sleep') as sleep,
+            self.assertRaisesRegex(RuntimeError, 'pending_confirmation'),
+        ):
+            plugins.install_plugins(
+                lambda args: {'plugins': [pending]},
+                directory,
+                manifest,
+                install=lambda entry: None,
+                readiness_timeout=10,
+            )
+        self.assertEqual(sleep.call_count, 1)
+
 
 if __name__ == '__main__':
     unittest.main()

web/tests/e2e/wizard-default-agent.spec.ts

@@ -58,6 +58,7 @@ for (const failsFirst of [false, true]) {
         plugin_author: 'langbot-team',
         plugin_name: 'LocalAgent',
         plugin_version: '0.1.7',
+        auto_confirm_permissions: true,
       });
       order.push('install');
       await route.fulfill({