feat(plugin): route certified installs to shared workers

src/langbot/pkg/plugin/certification.py

@@ -14,6 +14,7 @@ from enum import Enum
 from cryptography.exceptions import InvalidSignature
 from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PublicKey
 from langbot_plugin.certification import normalized_zip_digest, verify_archive
+from langbot_plugin.entities.io.context import PluginExecutionMode
 
 
 SHARED_RUNTIME_V1 = 'shared-runtime-v1'
@@ -232,3 +233,27 @@ def decide_plugin_log_visibility(facts: PluginCertificationFacts) -> PluginLogVi
     if facts.certificate.is_valid_shared_runtime:
         return PluginLogVisibility.TENANT_SCOPED
     return PluginLogVisibility.DETAILED_PROCESS
+
+
+def execution_mode_for_persisted_installation(
+    *,
+    artifact_digest: str,
+    install_info: object,
+) -> PluginExecutionMode:
+    """Derive placement only from persisted facts bound to the exact artifact."""
+
+    if not isinstance(install_info, Mapping):
+        return PluginExecutionMode.DEDICATED
+    certification = install_info.get('_certification')
+    if not isinstance(certification, Mapping):
+        return PluginExecutionMode.DEDICATED
+    shared_facts = {
+        'artifact_digest': artifact_digest,
+        'verification': CertificateVerification.VALID.value,
+        'certificate_runtime_profile': SHARED_RUNTIME_V1,
+        'runtime_profile': SHARED_RUNTIME_V1,
+        'admission_code': AdmissionCode.SHARED_ELIGIBLE.value,
+    }
+    if all(certification.get(key) == value for key, value in shared_facts.items()):
+        return PluginExecutionMode.SHARED_CERTIFIED
+    return PluginExecutionMode.DEDICATED

src/langbot/pkg/plugin/connector.py

@@ -34,6 +34,7 @@ from .certification import (
     PluginCertificationFacts,
     VerifiedArchiveCertificate,
     decide_plugin_admission,
+    execution_mode_for_persisted_installation,
     verify_plugin_archive_certificate,
 )
 from .github import (
@@ -62,6 +63,7 @@ from langbot_plugin.runtime.security import (
 )
 from langbot_plugin.entities.io.context import (
     InstallationBinding,
+    PluginExecutionMode,
     PluginInstallationDesiredState,
     PluginWorkerPolicy,
     RuntimeIdentity,
@@ -343,6 +345,13 @@ class PluginRuntimeConnector(ManagedRuntimeConnector):
             artifact_digest=setting.artifact_digest,
         )
 
+    @staticmethod
+    def _execution_mode_from_setting(setting: persistence_plugin.PluginSetting) -> PluginExecutionMode:
+        return execution_mode_for_persisted_installation(
+            artifact_digest=setting.artifact_digest,
+            install_info=setting.install_info,
+        )
+
     def _legacy_oss_bridge_binding(self, execution_context: ExecutionContext) -> InstallationBinding:
         seed = f'langbot:oss-plugin-bridge:{execution_context.instance_uuid}:{execution_context.workspace_uuid}'
         return InstallationBinding(
@@ -511,6 +520,7 @@ class PluginRuntimeConnector(ManagedRuntimeConnector):
                 PluginInstallationDesiredState(
                     binding=binding,
                     enabled=setting.enabled,
+                    execution_mode=self._execution_mode_from_setting(setting),
                 )
             )
         return tuple(desired_states)
@@ -526,6 +536,7 @@ class PluginRuntimeConnector(ManagedRuntimeConnector):
             desired.binding,
             artifact_package=artifact_package,
             enabled=desired.enabled,
+            execution_mode=desired.execution_mode,
         )
         self._raise_apply_failure(desired, result)
         if result.get('state') != 'artifact_missing':
@@ -551,6 +562,7 @@ class PluginRuntimeConnector(ManagedRuntimeConnector):
             desired.binding,
             artifact_package=persisted_package,
             enabled=desired.enabled,
+            execution_mode=desired.execution_mode,
         )
         self._raise_apply_failure(desired, repaired)
         if repaired.get('state') == 'artifact_missing':
@@ -1793,6 +1805,7 @@ class PluginRuntimeConnector(ManagedRuntimeConnector):
         }:
             raise ValueError(decision.code.value)
         certification_info = {
+            'artifact_digest': hashlib.sha256(file_bytes).hexdigest(),
             'normalized_digest': facts.artifact_digest,
             'verification': facts.certificate.verification.value,
             'certificate_runtime_profile': facts.certificate.runtime_profile,
@@ -1893,7 +1906,14 @@ class PluginRuntimeConnector(ManagedRuntimeConnector):
                     install_info=install_info,
                     artifact_digest=artifact_digest,
                 )
-                desired = PluginInstallationDesiredState(binding=binding, enabled=True)
+                desired = PluginInstallationDesiredState(
+                    binding=binding,
+                    enabled=True,
+                    execution_mode=execution_mode_for_persisted_installation(
+                        artifact_digest=artifact_digest,
+                        install_info=install_info,
+                    ),
+                )
                 runtime_handler.register_installation_binding(
                     binding,
                     plugin_author=plugin_author,
@@ -2143,6 +2163,7 @@ class PluginRuntimeConnector(ManagedRuntimeConnector):
         desired = PluginInstallationDesiredState(
             binding=binding,
             enabled=setting.enabled,
+            execution_mode=self._execution_mode_from_setting(setting),
         )
         is_legacy_oss = self.runtime_profile == 'oss_dev' and (
             not isinstance(setting.install_info, dict)

src/langbot/pkg/plugin/handler.py

@@ -24,6 +24,7 @@ from langbot_plugin.entities.io.context import (
     ActionContext,
     ApplyPluginInstallationRequest,
     InstallationBinding,
+    PluginExecutionMode,
     PluginInstallationDesiredState,
     PluginWorkerPolicy,
     ReconcilePluginInstallationsRequest,
@@ -2793,6 +2794,7 @@ class RuntimeConnectionHandler(handler.Handler):
         *,
         artifact_package: bytes | None,
         enabled: bool,
+        execution_mode: PluginExecutionMode = PluginExecutionMode.DEDICATED,
     ) -> dict[str, Any]:
         with self.installation_scope(binding):
             artifact_file_key = None
@@ -2801,6 +2803,7 @@ class RuntimeConnectionHandler(handler.Handler):
             request = ApplyPluginInstallationRequest(
                 artifact_file_key=artifact_file_key,
                 enabled=enabled,
+                execution_mode=execution_mode,
             )
             return await self.call_action(
                 LangBotToRuntimeAction.APPLY_PLUGIN_INSTALLATION,