feat(pipeline-migration): relax legacy config migration and pin latest runner plugins

- Coerce/default/drop invalid or unknown legacy fields with warnings instead of blocking - Never block on LocalAgent model; prune dangling KB/tool/MCP references - Mark custom (non-builtin) 4.10 runners as unsupported_runner instead of failing - Pin runner plugins to current marketplace versions (0.2.x); fall back to latest when a pin is withdrawn - Raise plugin install/upgrade timeout to 600s for slow dependency installs - Assistant configure_pipeline writes the 4.11 runner shape Co-Authored-By: Claude Opus 5.5 <[email protected]>
2026-11-03 22:32:27 +08:00 · 2026-10-14 00:09:52 +08:00
parent 0ed2a513f9
commit c0cfb5c18c
21 changed files with 1103 additions and 498 deletions
@@ -312,12 +312,35 @@ async def execute_tool(ap, context, name: str, arguments: dict):
            if await ap.knowledge_service.get_knowledge_base(context, kb_id) is None:
                raise ValueError('Knowledge base not found')
        config = copy.deepcopy(pipeline['config'])
-        config['ai']['runner']['runner'] = 'local-agent'
-        local = config['ai']['local-agent']
-        local['model']['primary'] = args['model_uuid']
+        runner_id = 'plugin:langbot-team/LocalAgent/default'
+        ai = config.get('ai') if isinstance(config.get('ai'), dict) else {}
+        runner = ai.get('runner') if isinstance(ai.get('runner'), dict) else {}
+        runner_configs = ai.get('runner_config') if isinstance(ai.get('runner_config'), dict) else {}
+        local = runner_configs.get(runner_id) if isinstance(runner_configs.get(runner_id), dict) else {}
+        model = local.get('model') if isinstance(local.get('model'), dict) else {}
+        local['model'] = {
+            'primary': args['model_uuid'],
+            'fallbacks': model.get('fallbacks') if isinstance(model.get('fallbacks'), list) else [],
+            'reasoning': model.get('reasoning') if isinstance(model.get('reasoning'), dict) else {},
+        }
        local['prompt'] = [{'role': 'system', 'content': args['system_prompt']}]
        local['knowledge-bases'] = args['knowledge_base_uuids']
-        await ap.pipeline_service.update_pipeline(context, args['pipeline_uuid'], {'config': config})
+        # 4.11 shape: ai.runner.id + ai.runner_config[id]; legacy keys are dropped.
+        ai['runner'] = {'id': runner_id, 'expire-time': runner.get('expire-time', 0)}
+        ai['runner_config'] = {**runner_configs, runner_id: local}
+        ai.pop('local-agent', None)
+        config['ai'] = ai
+        update = {'config': config}
+        prefs = pipeline.get('extensions_preferences')
+        if isinstance(prefs, dict) and prefs.get('enable_all_plugins') is False:
+            plugins = prefs.get('plugins') if isinstance(prefs.get('plugins'), list) else []
+            entry = {'author': 'langbot-team', 'name': 'LocalAgent'}
+            if not any(
+                isinstance(p, dict) and p.get('author') == entry['author'] and p.get('name') == entry['name']
+                for p in plugins
+            ):
+                update['extensions_preferences'] = {**prefs, 'plugins': [*plugins, entry]}
+        await ap.pipeline_service.update_pipeline(context, args['pipeline_uuid'], update)
        return {'uuid': args['pipeline_uuid'], 'url': f'/home/pipelines?id={args["pipeline_uuid"]}'}
    resource_id = await ap.knowledge_service.create_knowledge_base(context, args)
    return {'uuid': resource_id, 'url': f'/home/knowledge?id={resource_id}'}
@@ -152,6 +152,28 @@ def _json(value):
    return json.dumps(value, sort_keys=True, separators=(',', ':'), ensure_ascii=False, default=str)


+_SKIPPED_STATES = ('already_current', 'not_legacy', 'unsupported_runner')
+
+
+def _version_at_least(installed, minimum):
+    # Pins are reviewed minimums; never downgrade an installed newer plugin.
+    from packaging.version import InvalidVersion, Version
+
+    if installed == minimum:
+        return True
+    try:
+        return Version(str(installed)) >= Version(str(minimum))
+    except (InvalidVersion, TypeError):
+        return False
+
+
+def _candidate(row, plan):
+    candidate = {**row, 'config': plan.get('config')}
+    if plan.get('extensions_preferences') is not None:
+        candidate['extensions_preferences'] = plan['extensions_preferences']
+    return candidate
+
+
 def _fingerprint(row):
    # SQL timestamps are neither monotonic revisions nor portable CAS values.
    return hashlib.sha256(
@@ -312,7 +334,7 @@ class PipelineMigrationService:
                pending = dict(snapshots[0]._mapping)
                source = pending['source_snapshot']
                plan = plan_legacy_pipeline(source['config'], source['extensions_preferences'])
-                candidate = {**row, 'config': plan.get('config')}
+                candidate = _candidate(row, plan)
                if plan['state'] != 'ready' or _fingerprint(candidate) != _fingerprint(row):
                    raise MigrationError('planner_changed')
        facts = await self._plugin(ctx, plan)
@@ -439,7 +461,7 @@ class PipelineMigrationService:
                sources = []
                for row in await self._rows(ctx):
                    _, _, _, state, _ = await self._plan(ctx, row)
-                    if state not in ('already_current', 'not_legacy'):
+                    if state not in _SKIPPED_STATES:
                        sources.append(row)
            if not sources:
                raise MigrationError('nothing_to_migrate')
@@ -482,8 +504,11 @@ class PipelineMigrationService:
                        hard_blockers = [b for b in blockers if b['code'] not in ('plugin_missing', 'plugin_disabled')]
                        if hard_blockers:
                            raise MigrationError(hard_blockers[0]['code'])
-                        if state in ('already_current', 'not_legacy'):
-                            result.update(state='already_current', code=None)
+                        if state in _SKIPPED_STATES:
+                            result.update(
+                                state='unsupported_runner' if state == 'unsupported_runner' else 'already_current',
+                                code=None,
+                            )
                            continue
                        target = plan.get('target_plugin')
                        if not target:
@@ -505,7 +530,7 @@ class PipelineMigrationService:
                                facts
                                and facts['enabled']
                                and descriptor
-                                and descriptor.plugin_version == target['version']
+                                and _version_at_least(descriptor.plugin_version, target['version'])
                            )
                            if not ready:
                                task_context.metadata['phase'] = 'installing'
@@ -516,15 +541,22 @@ class PipelineMigrationService:
                                try:
                                    await self._authorize(ctx)
                                    await self.ap.plugin_connector.require_workspace_context(execution)
-                                    await self.ap.plugin_connector.install_plugin(
-                                        PluginInstallSource.MARKETPLACE,
-                                        {
-                                            'plugin_author': target['author'],
-                                            'plugin_name': target['name'],
-                                            'plugin_version': target['version'],
-                                        },
-                                        task_context=install_context,
-                                    )
+                                    spec = {'plugin_author': target['author'], 'plugin_name': target['name']}
+                                    try:
+                                        await self.ap.plugin_connector.install_plugin(
+                                            PluginInstallSource.MARKETPLACE,
+                                            {**spec, 'plugin_version': target['version']},
+                                            task_context=install_context,
+                                        )
+                                    except (MarketplacePluginVersionNotFoundError, httpx.HTTPStatusError) as exc:
+                                        if isinstance(exc, httpx.HTTPStatusError) and exc.response.status_code != 404:
+                                            raise
+                                        # The reviewed pin was withdrawn: fall back to the
+                                        # marketplace latest; _verify_runtime still checks
+                                        # the schema against the planned config.
+                                        await self.ap.plugin_connector.install_plugin(
+                                            PluginInstallSource.MARKETPLACE, spec, task_context=install_context
+                                        )
                                    install_code = None
                                except PluginRuntimeNotConnectedError:
                                    install_code = 'plugin_runtime_unavailable'
@@ -595,7 +627,7 @@ class PipelineMigrationService:
        descriptor = next((r for r in runners if r.id == plan['target_runner_id']), None)
        if descriptor is None or 'agent' not in descriptor.usages:
            raise MigrationError('runner_unavailable')
-        if descriptor.plugin_version != plan['target_plugin']['version']:
+        if not _version_at_least(descriptor.plugin_version, plan['target_plugin']['version']):
            raise MigrationError('plugin_version_incompatible')
        config = plan['config']
        RunnerConfigResolver.validate_pipeline_config(config)
@@ -685,12 +717,23 @@ class PipelineMigrationService:
        )

    async def _verify_resources(self, execution, descriptor, runner_config):
-        """Fail closed on references the scoped Host cannot authorize.
+        """Fail closed on unauthorized grants; prune dangling references.

        This runs only on explicit execution, outside the mutation transaction.
        Resource use is still reauthorized by the Host on every eventual run.
        """

+        async def has_row(model, identifier, *, column=None):
+            column = model.uuid if column is None else column
+            return (
+                await self.pm.execute_async(
+                    sa.select(model.uuid).where(
+                        model.workspace_uuid == execution.workspace_uuid,
+                        column == identifier,
+                    )
+                )
+            ).first() is not None
+
        async def require_row(model, identifier, *, column=None):
            column = model.uuid if column is None else column
            found = (
@@ -715,20 +758,26 @@ class PipelineMigrationService:
            if value and value not in config_schema.NONE_SENTINELS:
                await require_row(EmbeddingModel, value)

+        # Dangling knowledge base / tool / MCP references (deleted in 4.10)
+        # are pruned rather than blocking the migration.
        kb_ids = runner_config.get('knowledge-bases', [])
-        if not isinstance(kb_ids, list) or any(not isinstance(v, str) or not v for v in kb_ids):
-            raise MigrationError('runner_resource_unavailable')
+        if not isinstance(kb_ids, list):
+            kb_ids = []
+        kb_ids = [v for v in kb_ids if isinstance(v, str) and v and await has_row(KnowledgeBase, v)]
+        if 'knowledge-bases' in runner_config:
+            runner_config['knowledge-bases'] = kb_ids
        if kb_ids:
            if not config_schema.uses_host_knowledge_bases(descriptor) or 'retrieve' not in getattr(
                permissions, 'knowledge_bases', []
            ):
                raise MigrationError('runner_resource_unavailable')
-            for kb_id in kb_ids:
-                await require_row(KnowledgeBase, kb_id)

        tool_names = runner_config.get('tools', [])
-        if not isinstance(tool_names, list) or any(not isinstance(v, str) or not v for v in tool_names):
-            raise MigrationError('runner_resource_unavailable')
+        if not isinstance(tool_names, list):
+            tool_names = []
+        tool_names = [v for v in tool_names if isinstance(v, str) and v]
+        if 'tools' in runner_config:
+            runner_config['tools'] = tool_names
        tool_grant = (
            runner_config.get('enable-all-tools', False)
            or tool_names
@@ -745,20 +794,22 @@ class PipelineMigrationService:
                descriptor,
                runner_config,
            )
-            if not set(tool_names) <= {tool['tool_name'] for tool in tools}:
-                raise MigrationError('runner_resource_unavailable')
+            available = {tool['tool_name'] for tool in tools}
+            if 'tools' in runner_config:
+                runner_config['tools'] = [v for v in tool_names if v in available]

-        for attachment in runner_config.get('mcp-resources', []):
-            if not isinstance(attachment, dict):
-                raise MigrationError('runner_resource_unavailable')
-            identifier = attachment.get('server_uuid') or attachment.get('server_id')
-            name = attachment.get('server_name')
-            if identifier:
-                await require_row(MCPServer, identifier)
-            elif name:
-                await require_row(MCPServer, name, column=MCPServer.name)
-            else:
-                raise MigrationError('runner_resource_unavailable')
+        if 'mcp-resources' in runner_config:
+            kept = []
+            for attachment in runner_config.get('mcp-resources') or []:
+                if not isinstance(attachment, dict):
+                    continue
+                identifier = attachment.get('server_uuid') or attachment.get('server_id')
+                name = attachment.get('server_name')
+                if (identifier and await has_row(MCPServer, identifier)) or (
+                    not identifier and name and await has_row(MCPServer, name, column=MCPServer.name)
+                ):
+                    kept.append(attachment)
+            runner_config['mcp-resources'] = kept

    async def _lock(self, ctx, pipeline_uuid, session):
        # Every pending producer shares this row and revalidates its captured
@@ -792,11 +843,15 @@ class PipelineMigrationService:
            if pending:
                return pending['uuid'], pending['target_fingerprint']
            target = {**row, 'config': candidate['config']}
+            values = {'config': candidate['config']}
+            if plan.get('extensions_preferences') is not None:
+                target['extensions_preferences'] = plan['extensions_preferences']
+                values['extensions_preferences'] = plan['extensions_preferences']
            t = LegacyPipeline.__table__
            cas = [t.c.workspace_uuid == ctx.workspace_uuid, t.c.uuid == row['uuid']]
            for key in ('config', 'stages', 'extensions_preferences'):
                cas.append(sa.cast(t.c[key], sa.Text) == row['_raw_' + key])
-            result = await self.pm.execute_async(sa.update(t).where(*cas).values(config=candidate['config']))
+            result = await self.pm.execute_async(sa.update(t).where(*cas).values(**values))
            if result.rowcount != 1:
                raise MigrationError('preview_stale')
            snapshot_uuid = str(uuid.uuid4())
@@ -861,6 +916,8 @@ class PipelineMigrationService:
                    RunnerConfigResolver.validate_pipeline_config(plan['config'])
                    candidate_entity = {k: copy.deepcopy(v) for k, v in row.items() if not k.startswith('_')}
                    candidate_entity['config'] = copy.deepcopy(plan['config'])
+                    if plan.get('extensions_preferences') is not None:
+                        candidate_entity['extensions_preferences'] = copy.deepcopy(plan['extensions_preferences'])
                    runtime = await self.ap.pipeline_mgr.prepare_pipeline(execution, copy.deepcopy(candidate_entity))
                    if not data_only and await self._verify_runtime(execution, plan) != runtime_schema:
                        raise MigrationError('runner_schema_changed')