fix(plugin): keep certified marketplace packages installable on OSS

The certified-archive admission gate treated any declared certificate it could
not resolve as an untrusted archive and rejected the install with
CERTIFIED_PLUGIN_OSS_FORCE_REQUIRED. OSS ships an empty
plugin.certification.trusted_public_keys ring and the marketplace signs every
package with its own issuer key, so all certified marketplace packages (for
example langbot-team/RunnerDemo and langbot-team/LocalAgent) failed at the
'validating plugin package' step before artifact storage.

Admission now distinguishes an unresolvable declaration from a configured trust
decision that fails:

- record certificate_id only when the key_id is actually present in the
  configured ring, so an empty ring yields an unresolvable declaration;
- OSS degrades an unresolvable declaration to the existing oss_dev dedicated
  profile with CERTIFIED_PLUGIN_OSS_UNTRUSTED_DEDICATED instead of blocking;
- a resolvable declaration that still fails keeps requiring the explicit
  administrator force (CERTIFIED_PLUGIN_OSS_FORCE_REQUIRED);
- Cloud stays fail-closed and rejects before storage.

No shared-runtime privilege is granted when the issuer is not trusted, so this
withholds isolation rather than escalating it.

tests/integration/plugin/test_certified_plugin_admission.py

@@ -30,7 +30,7 @@ pytestmark = pytest.mark.integration
         ('cloud', 'signed_shared', False, 'shared-runtime-v1'),
         ('oss', 'signed_shared', False, 'shared-runtime-v1'),
         ('oss', 'legacy', False, 'dedicated'),
-        ('oss', 'invalid_shared', True, 'dedicated'),
+        ('oss', 'forged_shared', True, 'dedicated'),
     ],
 )
 async def test_install_plugin_admits_archive_before_persistence_and_applies_selected_profile(
@@ -81,7 +81,10 @@ async def test_cloud_rejects_untrusted_archive_before_storage_persistence_or_run
 
 @pytest.mark.asyncio
 async def test_oss_requires_explicit_administrator_force_for_declared_invalid_archive() -> None:
-    package, trusted_public_keys = _archive('invalid_shared')
+    # The archive declares the profile of a key this instance *does* trust but is
+    # signed by a different key, so the ring resolves its key_id and the failing
+    # signature is an explicit trust decision rather than an unconfigured ring.
+    package, trusted_public_keys = _archive('forged_shared')
     connector, _execution_context, _binding = _connector('oss', trusted_public_keys)
 
     with pytest.raises(ValueError, match='CERTIFIED_PLUGIN_OSS_FORCE_REQUIRED'):
@@ -92,6 +95,51 @@ async def test_oss_requires_explicit_administrator_force_for_declared_invalid_ar
     connector.handler.apply_plugin_installation.assert_not_awaited()
 
 
+@pytest.mark.asyncio
+async def test_oss_admits_unresolvable_declaration_on_the_dedicated_profile() -> None:
+    """A self-hosted instance without the issuer key ring must still install.
+
+    Certified marketplace packages declare ``shared-runtime-v1`` and are signed by
+    the marketplace issuer. An OSS instance that never configured
+    ``plugin.certification.trusted_public_keys`` cannot resolve that issuer, so it
+    must degrade the install to the dedicated profile instead of rejecting every
+    certified package with ``CERTIFIED_PLUGIN_OSS_FORCE_REQUIRED``.
+    """
+
+    package, _trusted_public_keys = _archive('signed_shared')
+    connector, execution_context, binding = _connector('oss', {})
+
+    await connector.install_plugin(PluginInstallSource.LOCAL, {'plugin_file': package})
+
+    persisted_info = connector._persist_installation_package.await_args.kwargs['install_info']
+    assert persisted_info['_certification']['runtime_profile'] == 'dedicated'
+    assert persisted_info['_certification']['admission_code'] == 'CERTIFIED_PLUGIN_OSS_UNTRUSTED_DEDICATED'
+    assert persisted_info['_certification']['verification'] == 'invalid'
+    connector._store_artifact_package.assert_awaited_once_with(
+        execution_context,
+        hashlib.sha256(package).hexdigest(),
+        package,
+    )
+    connector.handler.apply_plugin_installation.assert_awaited_once_with(
+        binding,
+        artifact_package=package,
+        enabled=True,
+    )
+
+
+@pytest.mark.asyncio
+async def test_cloud_rejects_unresolvable_declaration_before_storage() -> None:
+    package, _trusted_public_keys = _archive('signed_shared')
+    connector, _execution_context, _binding = _connector('cloud', {})
+
+    with pytest.raises(ValueError, match='CERTIFIED_PLUGIN_CLOUD_CERTIFICATE_INVALID'):
+        await connector.install_plugin(PluginInstallSource.LOCAL, {'plugin_file': package})
+
+    connector._store_artifact_package.assert_not_awaited()
+    connector._persist_installation_package.assert_not_awaited()
+    connector.handler.apply_plugin_installation.assert_not_awaited()
+
+
 @pytest.mark.asyncio
 @pytest.mark.parametrize('requested_version', [None, '1.0.0'])
 @pytest.mark.parametrize('archive_kind', ['signed_shared', 'legacy'])
@@ -201,6 +249,24 @@ def _archive(kind: str) -> tuple[bytes, dict[str, str]]:
 
     from langbot_plugin.certification import create_envelope, write_envelope
 
+    def _raw_public_key(private_key: Ed25519PrivateKey) -> str:
+        return base64.b64encode(
+            private_key.public_key().public_bytes(
+                serialization.Encoding.Raw,
+                serialization.PublicFormat.Raw,
+            )
+        ).decode('ascii')
+
+    if kind == 'forged_shared':
+        # Declares a key ID the instance trusts but signs with a different key,
+        # so the configured ring resolves the identity and rejects the signature.
+        trusted_key = Ed25519PrivateKey.generate()
+        forged_archive = write_envelope(
+            raw_archive,
+            create_envelope(raw_archive, 'trusted', Ed25519PrivateKey.generate().sign),
+        )
+        return forged_archive, {'trusted': _raw_public_key(trusted_key)}
+
     signing_key = Ed25519PrivateKey.generate()
     signed_archive = write_envelope(
         raw_archive,
@@ -210,11 +276,7 @@ def _archive(kind: str) -> tuple[bytes, dict[str, str]]:
             signing_key.sign,
         ),
     )
-    trusted_key = signing_key.public_key().public_bytes(
-        serialization.Encoding.Raw,
-        serialization.PublicFormat.Raw,
-    )
-    return signed_archive, {'ephemeral': base64.b64encode(trusted_key).decode('ascii')}
+    return signed_archive, {'ephemeral': _raw_public_key(signing_key)}
 
 
 def _normalized_digest(archive: bytes) -> str:

tests/unit_tests/plugin/test_certified_plugin_policy.py

@@ -8,27 +8,63 @@ import pytest
 
 
 @pytest.mark.parametrize(
-    ('deployment', 'certificate', 'force', 'expected_disposition', 'expected_code'),
+    ('deployment', 'certificate', 'certificate_id', 'force', 'expected_disposition', 'expected_code'),
     [
-        ('cloud', ('valid', 'shared-runtime-v1'), False, 'shared_eligible', 'CERTIFIED_PLUGIN_SHARED_ELIGIBLE'),
-        ('cloud', ('absent', None), False, 'rejected', 'CERTIFIED_PLUGIN_CLOUD_CERTIFICATE_REQUIRED'),
-        ('cloud', ('malformed', None), False, 'rejected', 'CERTIFIED_PLUGIN_CLOUD_CERTIFICATE_INVALID'),
-        ('cloud', ('invalid', 'shared-runtime-v1'), True, 'rejected', 'CERTIFIED_PLUGIN_CLOUD_CERTIFICATE_INVALID'),
-        ('oss', ('absent', None), False, 'dedicated_allowed', 'CERTIFIED_PLUGIN_OSS_LEGACY_DEDICATED'),
-        ('oss', ('valid', 'shared-runtime-v1'), False, 'shared_eligible', 'CERTIFIED_PLUGIN_SHARED_ELIGIBLE'),
+        ('cloud', ('valid', 'shared-runtime-v1'), 'issuer', False, 'shared_eligible', 'CERTIFIED_PLUGIN_SHARED_ELIGIBLE'),
+        ('cloud', ('absent', None), None, False, 'rejected', 'CERTIFIED_PLUGIN_CLOUD_CERTIFICATE_REQUIRED'),
+        ('cloud', ('malformed', None), None, False, 'rejected', 'CERTIFIED_PLUGIN_CLOUD_CERTIFICATE_INVALID'),
+        (
+            'cloud',
+            ('invalid', 'shared-runtime-v1'),
+            'issuer',
+            True,
+            'rejected',
+            'CERTIFIED_PLUGIN_CLOUD_CERTIFICATE_INVALID',
+        ),
+        ('oss', ('absent', None), None, False, 'dedicated_allowed', 'CERTIFIED_PLUGIN_OSS_LEGACY_DEDICATED'),
+        ('oss', ('valid', 'shared-runtime-v1'), 'issuer', False, 'shared_eligible', 'CERTIFIED_PLUGIN_SHARED_ELIGIBLE'),
         (
             'oss',
             ('invalid', 'shared-runtime-v1'),
+            'issuer',
             False,
             'administrator_force_required',
             'CERTIFIED_PLUGIN_OSS_FORCE_REQUIRED',
         ),
-        ('oss', ('invalid', 'shared-runtime-v1'), True, 'dedicated_allowed', 'CERTIFIED_PLUGIN_OSS_FORCED_DEDICATED'),
+        (
+            'oss',
+            ('invalid', 'shared-runtime-v1'),
+            'issuer',
+            True,
+            'dedicated_allowed',
+            'CERTIFIED_PLUGIN_OSS_FORCED_DEDICATED',
+        ),
+        # A declaration the operator cannot resolve (no trusted key ring configured)
+        # must keep the OSS install working on the dedicated profile instead of
+        # blocking every certified marketplace package.
+        (
+            'oss',
+            ('invalid', 'shared-runtime-v1'),
+            None,
+            False,
+            'dedicated_allowed',
+            'CERTIFIED_PLUGIN_OSS_UNTRUSTED_DEDICATED',
+        ),
+        (
+            'oss',
+            ('invalid', 'shared-runtime-v1'),
+            None,
+            True,
+            'dedicated_allowed',
+            'CERTIFIED_PLUGIN_OSS_UNTRUSTED_DEDICATED',
+        ),
+        ('oss', ('malformed', None), None, False, 'dedicated_allowed', 'CERTIFIED_PLUGIN_OSS_UNTRUSTED_DEDICATED'),
     ],
 )
 def test_admission_policy_enforces_certification_matrix(
     deployment: str,
     certificate: tuple[str, str | None],
+    certificate_id: str | None,
     force: bool,
     expected_disposition: str,
     expected_code: str,
@@ -47,6 +83,7 @@ def test_admission_policy_enforces_certification_matrix(
         certificate=CertificateFacts(
             verification=CertificateVerification(verification),
             runtime_profile=runtime_profile,
+            certificate_id=certificate_id,
         ),
     )