Merge origin/master into feat/marketplace-installed-state-and-search

Resolved one conflict in src/langbot/pkg/plugin/connector.py. Master added
archive admission (`_admit_plugin_archive` + the certification digest check
before Runtime apply) at the same points this branch added install-stage
reporting, so both sides are kept:

- the archive is admitted first, then the 'inspecting plugin package' stage is
  reported
- the certification digest is verified before Runtime apply, then the
  'installing or starting plugin' stage is reported

Verified after merge: py_compile + ruff, tsc --noEmit, 98/98 unit tests.

src/langbot/pkg/plugin/connector.py

@@ -22,6 +22,13 @@ from langbot_plugin.api.entities.builtin.pipeline.query import provider_session
 from ..core import app
 from . import handler
 from .archive import inspect_plugin_archive_metadata
+from .certification import (
+    AdmissionDisposition,
+    PluginCertificationFacts,
+    VerifiedArchiveCertificate,
+    decide_plugin_admission,
+    verify_plugin_archive_certificate,
+)
 from .github import (
     validate_github_plugin_install_info,
     validate_github_release_asset_url,
@@ -150,6 +157,30 @@ def _decode_json_object(body: bytes, *, subject: str) -> dict[str, Any]:
     return payload
 
 
+def _select_marketplace_plugin_version(
+    versions: Any,
+    *,
+    requested_version: str | None,
+    plugin_author: str,
+    plugin_name: str,
+) -> str:
+    if not isinstance(versions, list) or not versions:
+        raise ValueError(f'Plugin {plugin_author}/{plugin_name} has no versions')
+
+    if requested_version is None:
+        candidate = versions[0]
+        if not isinstance(candidate, dict) or not candidate.get('version'):
+            raise ValueError(f'Plugin {plugin_author}/{plugin_name} has no versions')
+        return str(candidate['version'])
+
+    for candidate in versions:
+        if isinstance(candidate, dict) and str(candidate.get('version') or '') == requested_version:
+            return requested_version
+    raise ValueError(
+        f'Plugin {plugin_author}/{plugin_name} version {requested_version} is not available in marketplace'
+    )
+
+
 class PluginRuntimeNotConnectedError(RuntimeError):
     """Raised when plugin runtime operations are requested before connection."""
 
@@ -1622,6 +1653,7 @@ class PluginRuntimeConnector(ManagedRuntimeConnector):
         execution_context: ExecutionContext,
         plugin_author: str,
         plugin_name: str,
+        plugin_version: str | None,
         task_context: taskmgr.TaskContext | None,
     ) -> tuple[bytes | None, str | None]:
         """Return a plugin package, or install an MCP/skill and return none."""
@@ -1686,21 +1718,59 @@ class PluginRuntimeConnector(ManagedRuntimeConnector):
                 subject='Marketplace plugin versions',
             )
             versions = versions_payload.get('data', {}).get('versions', [])
-            if (
-                not isinstance(versions, list)
-                or not versions
-                or not isinstance(versions[0], dict)
-                or not versions[0].get('version')
-            ):
-                raise ValueError(f'Plugin {plugin_author}/{plugin_name} has no versions')
-            latest_version = str(versions[0]['version'])
+            requested_version = str(plugin_version or '').strip()
+            version = _select_marketplace_plugin_version(
+                versions,
+                requested_version=requested_version or None,
+                plugin_author=plugin_author,
+                plugin_name=plugin_name,
+            )
             _download_status, plugin_package = await _marketplace_get(
                 client,
-                f'{space_url}/api/v1/marketplace/plugins/download/{plugin_author}/{plugin_name}/{latest_version}',
+                f'{space_url}/api/v1/marketplace/plugins/download/{plugin_author}/{plugin_name}/{version}',
                 max_bytes=_MARKETPLACE_PLUGIN_DOWNLOAD_MAX_BYTES,
                 task_context=task_context,
             )
-            return plugin_package, latest_version
+            return plugin_package, version
+
+    def _admit_plugin_archive(
+        self,
+        file_bytes: bytes,
+        install_info: dict[str, Any],
+    ) -> tuple[dict[str, Any], VerifiedArchiveCertificate]:
+        """Verify and admit one archive before it can reach durable storage or Runtime."""
+
+        certification_config = self.ap.instance_config.data.get('plugin', {}).get('certification', {})
+        if not isinstance(certification_config, dict):
+            raise ValueError('plugin.certification must be a mapping')
+        verified = verify_plugin_archive_certificate(
+            file_bytes,
+            trusted_public_keys=certification_config.get('trusted_public_keys', {}),
+        )
+        facts = PluginCertificationFacts(
+            installation_uuid='pending-installation',
+            artifact_digest=verified.normalized_digest,
+            certificate=verified.certificate,
+        )
+        decision = decide_plugin_admission(
+            deployment=getattr(getattr(self.ap, 'deployment', None), 'mode', 'oss'),
+            facts=facts,
+            administrator_force=install_info.get('administrator_force') is True,
+        )
+        if decision.disposition not in {
+            AdmissionDisposition.DEDICATED_ALLOWED,
+            AdmissionDisposition.SHARED_ELIGIBLE,
+        }:
+            raise ValueError(decision.code.value)
+        certification_info = {
+            'normalized_digest': facts.artifact_digest,
+            'verification': facts.certificate.verification.value,
+            'certificate_runtime_profile': facts.certificate.runtime_profile,
+            'certificate_id': facts.certificate.certificate_id,
+            'runtime_profile': decision.runtime_profile,
+            'admission_code': decision.code.value,
+        }
+        return {**install_info, '_certification': certification_info}, verified
 
     async def install_plugin(
         self,
@@ -1733,6 +1803,7 @@ class PluginRuntimeConnector(ManagedRuntimeConnector):
                 execution_context,
                 plugin_author,
                 plugin_name,
+                str(install_info.get('plugin_version') or '') or None,
                 task_context,
             )
             if file_bytes is None:
@@ -1752,6 +1823,7 @@ class PluginRuntimeConnector(ManagedRuntimeConnector):
         else:
             raise ValueError(f'Unsupported plugin install source: {install_source.value}')
 
+        install_info, verified_certificate = self._admit_plugin_archive(file_bytes, install_info)
         if task_context is not None:
             task_context.set_current_action('inspecting plugin package')
         manifest_author, manifest_name = self._inspect_plugin_package(file_bytes, task_context)
@@ -1788,6 +1860,9 @@ class PluginRuntimeConnector(ManagedRuntimeConnector):
             plugin_author=plugin_author,
             plugin_name=plugin_name,
         )
+        certification_facts = verified_certificate.for_installation(binding.installation_uuid)
+        if certification_facts.artifact_digest != install_info['_certification']['normalized_digest']:
+            raise RuntimeError('Plugin certification digest changed before Runtime apply')
         if task_context is not None:
             # The runtime installs the plugin's dependencies and starts it
             # inside apply_plugin_installation. It does not stream