test(tools): resolve and install latest published plugins with replay evidence (#2791)

tools/published-plugins.md

@@ -0,0 +1,52 @@
+# Published plugins in regression runs
+
+`regression_ci.py` and `regression_job.py` default to the latest published
+LocalAgent, LangRAG and GeneralParsers packages from the official LangBot
+marketplace. Plugin preparation happens before business testing. Each run reads
+`latest_version` once for each plugin, downloads that exact version, and fails
+if metadata, downloads, package identity, installation or initialization fail.
+There is no fallback to older releases, cached packages or a source checkout.
+
+The source runner needs the normal Core/SDK dependencies, Docker and `lbctl`:
+
+```sh
+python tools/regression_ci.py --lbctl /path/to/lbctl --output /path/to/new-run
+```
+
+The released deployment runner uses its existing job profile:
+
+```sh
+python tools/regression_job.py run --profile tools/regression_profiles/job.json --output /path/to/new-run
+```
+
+The job's default `plugin_mode` is `published`; its suite's `plugins` entries
+are replaced by the published preparation. Set `"plugin_mode": "profile"` in
+the job JSON to explicitly use the suite's existing local/GitHub/marketplace
+inputs. For source development, `regression_ci.py --plugins /path/to/langbot-plugins`
+retains packaging from a source checkout.
+
+Each published run keeps `plugins/manifest.json` and the downloaded ZIPs under
+its output directory. The manifest records the official metadata/download URLs,
+selected versions, SHA-256 hashes and installation readback (actual version,
+enabled state and initialized status). Archive identity/version must match the
+release metadata; the CLI must report that same version as enabled and initialized.
+Dependency or SDK incompatibility is a preparation failure and is not downgraded.
+
+To replay an earlier run, pass its manifest to the source runner:
+
+```sh
+python tools/regression_ci.py --lbctl /path/to/lbctl --output /path/to/replay --plugin-manifest /path/to/earlier/plugins/manifest.json
+```
+
+For the deployment runner, set `"plugin_manifest": "/path/to/earlier/plugins/manifest.json"`
+in the job JSON. Relative paths are resolved from the job profile directory.
+Replay downloads the recorded versions from the official marketplace without
+querying latest metadata and requires the recorded hashes. Missing or changed
+packages fail; retain the manifest and ZIPs as evidence. Supplied URLs cannot
+redirect replay to another registry or add other plugin identities.
+
+Installation uses `lbctl plugin install local --file <downloaded-ZIP> --wait`
+so the installed bytes are the ones whose hashes were recorded. This exercises
+published package compatibility and initialization; it does not exercise the
+marketplace install/upgrade endpoint itself. Workflow callers must omit their
+explicit `--plugins` argument to select the default published mode.

tools/regression_ci.py

@@ -15,6 +15,7 @@ import time
 import yaml
 
 from regression_gate import supervise
+from regression_plugins import install_published_plugins, prepare_published_plugins
 from regression_suite import Suite
 from regression_traffic import owned_box_containers, read_endpoint_snapshot
 
@@ -68,6 +69,9 @@ def run(args):
         return process.pid
 
     try:
+        published = None
+        if args.plugins is None:
+            published = prepare_published_plugins(root / 'plugins', args.plugin_manifest)
         config_path = create_minimal_config(private, port=15301)
         create_test_directories(private)
         config = yaml.safe_load(config_path.read_text())
@@ -171,12 +175,15 @@ def run(args):
         cli_data = yaml.safe_load(cli_config.read_text())
         cli_data['contexts']['smoke']['expected_workspace_uuid'] = identity['workspace_uuid']
         cli_config.write_text(yaml.safe_dump(cli_data))
-        for directory in ('Runner/LocalAgent', 'KnowledgeEngine/LangRAG', 'misc/GeneralParsers'):
-            source = args.plugins / directory
-            package_source = private / source.name
-            shutil.copytree(source, package_source, ignore=shutil.ignore_patterns('.git', '.venv*', '__pycache__'))
-            archive = shutil.make_archive(str(private / source.name), 'zip', root_dir=package_source)
-            suite.call(['plugin', 'install', 'local', '--file', archive, '--wait'])
+        if published is not None:
+            install_published_plugins(suite.call, root / 'plugins', published)
+        else:
+            for directory in ('Runner/LocalAgent', 'KnowledgeEngine/LangRAG', 'misc/GeneralParsers'):
+                source = args.plugins / directory
+                package_source = private / source.name
+                shutil.copytree(source, package_source, ignore=shutil.ignore_patterns('.git', '.venv*', '__pycache__'))
+                archive = shutil.make_archive(str(private / source.name), 'zip', root_dir=package_source)
+                suite.call(['plugin', 'install', 'local', '--file', archive, '--wait'])
         generated = private / 'smoke.json'
         generated.write_text(json.dumps(profile, indent=2))
         return supervise(
@@ -214,7 +221,9 @@ def main():
     parser = argparse.ArgumentParser(description=__doc__)
     parser.add_argument('--output', type=Path, required=True)
     parser.add_argument('--lbctl', type=Path, required=True)
-    parser.add_argument('--plugins', type=Path, required=True)
+    plugins = parser.add_mutually_exclusive_group()
+    plugins.add_argument('--plugins', type=Path, help='Explicit developer override: package a plugin source checkout')
+    plugins.add_argument('--plugin-manifest', type=Path, help='Replay published versions and hashes from a prior run')
     args = parser.parse_args()
     if args.output.exists():
         parser.error('--output must be a new directory')

tools/regression_job.py

@@ -17,6 +17,7 @@ import urllib.request
 import yaml
 
 from regression_suite import Suite, number, resolve, validate
+from regression_plugins import install_published_plugins, prepare_published_plugins
 from regression_traffic import owned_box_containers
 
 
@@ -34,8 +35,30 @@ def load_profile(path):
     profile = json.loads(path.read_text())
     if not isinstance(profile, dict) or profile.get('version') != 1:
         raise ValueError('job profile version must be 1')
-    if profile.keys() - {'version', 'lbctl', 'suite', 'deployment', 'vars', 'cleanup', 'purge_data', 'mock_image'}:
+    if profile.keys() - {
+        'version',
+        'lbctl',
+        'suite',
+        'deployment',
+        'vars',
+        'cleanup',
+        'purge_data',
+        'mock_image',
+        'plugin_mode',
+        'plugin_manifest',
+    }:
         raise ValueError('unknown job profile field')
+    if profile.get('plugin_mode', 'published') not in ('published', 'profile'):
+        raise ValueError('plugin_mode must be published/profile')
+    if 'plugin_manifest' in profile:
+        if profile.get('plugin_mode', 'published') != 'published':
+            raise ValueError('plugin_manifest requires published plugin_mode')
+        if not isinstance(profile['plugin_manifest'], str) or not profile['plugin_manifest']:
+            raise ValueError('plugin_manifest must name a prior published manifest')
+        manifest_path = Path(profile['plugin_manifest']).expanduser()
+        if not manifest_path.is_absolute():
+            manifest_path = path.parent / manifest_path
+        profile['plugin_manifest'] = str(manifest_path.resolve())
     deployment = profile.get('deployment', {})
     if not isinstance(deployment, dict) or deployment.keys() - {'version', 'port'}:
         raise ValueError('deployment accepts version and port')
@@ -52,6 +75,8 @@ def load_profile(path):
     suite_path = Path(profile['suite']).expanduser()
     suite_path = suite_path if suite_path.is_absolute() else path.parent / suite_path
     suite = json.loads(suite_path.read_text())
+    if profile.get('plugin_mode', 'published') == 'published':
+        suite['plugins'] = []
     suite.setdefault('vars', {}).update(profile.get('vars', {}))
     suite['vars'].update(
         mock_base_url='http://regression_mock:18080/v1', embedding_base_url='http://regression_mock:18080/v1'
@@ -198,6 +223,12 @@ class Job:
         raise RuntimeError(f'readiness deadline exceeded: {url}')
 
     def prepare(self):
+        published = None
+        if self.profile.get('plugin_mode', 'published') == 'published':
+            self.state['phase'] = 'plugins'
+            self.save()
+            published = prepare_published_plugins(self.output / 'plugins', self.profile.get('plugin_manifest'))
+            self.suite_profile['plugins'] = []
         variables = {
             **self.suite_profile.get('vars', {}),
             'profile_dir': str(self.suite_path.parent),
@@ -324,6 +355,12 @@ class Job:
         destination.mkdir(mode=0o700)
         suite = Suite(profile, self.suite_path, destination, env=self.env)
         self.redactions.extend(suite.secrets)
+        if published is not None:
+            self.state['phase'] = 'plugins'
+            self.save()
+            install_published_plugins(suite.call, self.output / 'plugins', published)
+            self.state['phase'] = 'resources'
+            self.save()
         suite.write('profile.json', profile)
         if 'smoke' not in profile:
             from regression_analysis import workload_contract

tools/regression_plugins.py

@@ -0,0 +1,170 @@
+"""Resolve official published regression plugins once, then install the recorded bytes."""
+
+from datetime import datetime, timezone
+import hashlib
+import io
+import json
+from pathlib import Path
+import re
+import urllib.error
+import urllib.request
+import zipfile
+
+import yaml
+
+
+MARKETPLACE = 'https://space.langbot.app'
+PLUGIN_IDS = ('langbot-team/LocalAgent', 'langbot-team/LangRAG', 'langbot-team/GeneralParsers')
+METADATA_LIMIT = 1024 * 1024
+PACKAGE_LIMIT = 64 * 1024 * 1024
+
+
+def write_manifest(directory, manifest):
+    path = directory / 'manifest.json'
+    temporary = directory / 'manifest.json.tmp'
+    temporary.write_text(json.dumps(manifest, ensure_ascii=False, indent=2) + '\n')
+    temporary.replace(path)
+
+
+def fetch(url, limit):
+    request = urllib.request.Request(url, headers={'User-Agent': 'LangBot-regression/1'})
+    try:
+        with urllib.request.urlopen(request, timeout=60) as response:
+            if response.status != 200:
+                raise RuntimeError(f'Marketplace request failed: HTTP {response.status}: {url}')
+            content = response.read(limit + 1)
+    except (urllib.error.URLError, OSError) as exc:
+        raise RuntimeError(f'Marketplace request failed: {url}: {exc}') from exc
+    if len(content) > limit:
+        raise ValueError(f'Marketplace response exceeds {limit} bytes: {url}')
+    return content
+
+
+def release_version(value):
+    if not isinstance(value, str) or not re.fullmatch(r'[a-zA-Z0-9][a-zA-Z0-9._+-]{0,127}', value):
+        raise ValueError(f'Invalid published plugin version: {value!r}')
+    return value
+
+
+def replay_versions(path):
+    manifest = json.loads(Path(path).read_text())
+    if not isinstance(manifest, dict) or manifest.get('version') != 1 or manifest.get('marketplace') != MARKETPLACE:
+        raise ValueError('Replay requires a version 1 official marketplace manifest')
+    entries = manifest.get('plugins')
+    if not isinstance(entries, list) or len(entries) != len(PLUGIN_IDS):
+        raise ValueError('Replay manifest must contain the three regression plugins')
+    selected = {}
+    for entry in entries:
+        if not isinstance(entry, dict) or entry.get('id') not in PLUGIN_IDS or entry['id'] in selected:
+            raise ValueError('Replay manifest has an unknown or duplicate plugin')
+        version = release_version(entry.get('version'))
+        digest = entry.get('sha256')
+        if not isinstance(digest, str) or not re.fullmatch('[0-9a-f]{64}', digest):
+            raise ValueError('Replay manifest requires a package sha256 for every plugin')
+        selected[entry['id']] = (version, digest)
+    return selected
+
+
+def check_archive(content, plugin_id, version):
+    try:
+        with zipfile.ZipFile(io.BytesIO(content)) as archive:
+            if archive.namelist().count('manifest.yaml') != 1:
+                raise ValueError('package needs exactly one root manifest.yaml')
+            info = archive.getinfo('manifest.yaml')
+            if info.file_size > METADATA_LIMIT:
+                raise ValueError('package manifest exceeds the metadata limit')
+            manifest = yaml.safe_load(archive.read(info))
+        metadata = manifest.get('metadata') if isinstance(manifest, dict) else None
+        if (
+            not isinstance(metadata, dict)
+            or manifest.get('kind') != 'Plugin'
+            or f'{metadata.get("author")}/{metadata.get("name")}' != plugin_id
+            or metadata.get('version') != version
+        ):
+            raise ValueError('package manifest identity/version differs from the selected release')
+    except (zipfile.BadZipFile, yaml.YAMLError, KeyError, ValueError, RuntimeError) as exc:
+        raise ValueError(f'Invalid published package for {plugin_id}@{version}: {exc}') from exc
+
+
+def prepare_published_plugins(directory, replay=None):
+    """Download releases; replay reads only identities, versions and hashes, never supplied URLs."""
+    directory = Path(directory)
+    directory.mkdir(parents=True, exist_ok=False)
+    selected = replay_versions(replay) if replay is not None else None
+    manifest = {
+        'version': 1,
+        'marketplace': MARKETPLACE,
+        'resolved_at': datetime.now(timezone.utc).isoformat(),
+        'selection': 'replay' if selected is not None else 'latest',
+        'plugins': [],
+    }
+    write_manifest(directory, manifest)
+    for plugin_id in PLUGIN_IDS:
+        metadata_url = f'{MARKETPLACE}/api/v1/marketplace/plugins/{plugin_id}'
+        if selected is None:
+            payload = json.loads(fetch(metadata_url, METADATA_LIMIT))
+            data = payload.get('data') if isinstance(payload, dict) else None
+            plugin = data.get('plugin') if isinstance(data, dict) else None
+            if (
+                not isinstance(payload, dict)
+                or payload.get('code') != 0
+                or not isinstance(plugin, dict)
+                or plugin.get('plugin_id') != plugin_id
+                or f'{plugin.get("author")}/{plugin.get("name")}' != plugin_id
+                or plugin.get('status') != 'live'
+                or plugin.get('deleted_at') is not None
+            ):
+                raise ValueError(f'Marketplace did not return a live published plugin: {plugin_id}')
+            version = release_version(plugin.get('latest_version'))
+        else:
+            version = selected[plugin_id][0]
+        manifest['plugins'].append(
+            {
+                'id': plugin_id,
+                'version': version,
+                'metadata_url': metadata_url,
+                'package_url': f'{MARKETPLACE}/api/v1/marketplace/plugins/download/{plugin_id}/{version}',
+                'archive': f'{plugin_id.split("/")[1]}-{version}.zip',
+            }
+        )
+        write_manifest(directory, manifest)
+
+    for entry in manifest['plugins']:
+        content = fetch(entry['package_url'], PACKAGE_LIMIT)
+        digest = hashlib.sha256(content).hexdigest()
+        if selected is not None and digest != selected[entry['id']][1]:
+            raise ValueError(f'Replay package sha256 changed: {entry["id"]}@{entry["version"]}')
+        check_archive(content, entry['id'], entry['version'])
+        (directory / entry['archive']).write_bytes(content)
+        entry['sha256'] = digest
+        write_manifest(directory, manifest)
+    return manifest
+
+
+def install_published_plugins(call, directory, manifest):
+    """Install the downloaded packages and require exact, initialized CLI readback."""
+    directory = Path(directory)
+    for entry in manifest['plugins']:
+        archive = directory / entry['archive']
+        if hashlib.sha256(archive.read_bytes()).hexdigest() != entry['sha256']:
+            raise ValueError(f'Plugin archive changed before installation: {entry["id"]}')
+        try:
+            call(['plugin', 'install', 'local', '--file', str(archive), '--wait'])
+        except RuntimeError as exc:
+            raise RuntimeError(f'Published 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']]
+        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'
+        ):
+            raise RuntimeError(
+                f'Published plugin is not ready at the selected version: {entry["id"]}: {entry["installed"]}'
+            )

tools/test_regression_job.py

@@ -13,7 +13,7 @@ from unittest.mock import patch
 
 import yaml
 
-from regression_job import Job, main
+from regression_job import Job, load_profile, main
 from test_regression_suite import FAKE_CLI, profile
 
 
@@ -89,6 +89,36 @@ class ControlledJob(Job):
 
 
 class JobTests(unittest.TestCase):
+    def test_published_resolution_failure_is_preparation_failure_without_deployment(self):
+        with tempfile.TemporaryDirectory() as folder:
+            root = Path(folder)
+            job = ControlledJob({}, profile(), root / 'profile.json', root / 'run', 'unused')
+            with patch('regression_job.prepare_published_plugins', side_effect=RuntimeError('release unavailable')):
+                result = job.run()
+            self.assertFalse(result['passed'])
+            self.assertFalse(result['completed'])
+            self.assertEqual(result['phase'], 'plugins')
+            self.assertEqual(result['errors'], ['release unavailable'])
+            self.assertEqual(job.calls, [])
+
+    def test_job_defaults_to_published_plugins_and_keeps_explicit_profile_mode(self):
+        with tempfile.TemporaryDirectory() as folder:
+            root = Path(folder)
+            suite = profile()
+            suite['plugins'] = [{'source': 'local', 'file': 'developer.zip'}]
+            (root / 'suite.json').write_text(json.dumps(suite))
+            path = root / 'job.json'
+            manifest = {'version': 1, 'suite': 'suite.json', 'deployment': {'version': 'v4.11.0'}}
+            path.write_text(json.dumps(manifest))
+            self.assertEqual(load_profile(path)[1]['plugins'], [])
+            manifest['plugin_mode'] = 'profile'
+            path.write_text(json.dumps(manifest))
+            self.assertEqual(load_profile(path)[1]['plugins'], suite['plugins'])
+            manifest['plugin_manifest'] = 'previous/manifest.json'
+            path.write_text(json.dumps(manifest))
+            with self.assertRaisesRegex(ValueError, 'requires published'):
+                load_profile(path)
+
     def test_lifecycle_success_and_partial_install_recovery(self):
         class Health(BaseHTTPRequestHandler):
             def do_GET(self):
@@ -108,7 +138,11 @@ class JobTests(unittest.TestCase):
                 cli = root / 'lbctl-test'
                 cli.write_text(f'#!{sys.executable}\n' + FAKE_CLI)
                 cli.chmod(0o700)
-                manifest = {'lbctl': str(cli), 'deployment': {'version': 'v4.11.0', 'port': 15351}}
+                manifest = {
+                    'lbctl': str(cli),
+                    'deployment': {'version': 'v4.11.0', 'port': 15351},
+                    'plugin_mode': 'profile',
+                }
                 suite = profile()
                 suite['resources'] = [{'id': 'chat', 'kind': 'pipeline', 'body': {'name': 'QA', 'config': {'ai': {}}}}]
                 suite['cases'][0].update(args=['pipeline', 'run', '${chat}'], expect='TEST_OK')

tools/test_regression_plugins.py

@@ -0,0 +1,192 @@
+"""Exercise published resolution, replay integrity and installation readback."""
+
+from collections import Counter
+import copy
+from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
+import io
+import json
+from pathlib import Path
+import tempfile
+import threading
+import unittest
+from unittest.mock import patch
+import zipfile
+
+import regression_plugins as plugins
+
+
+def archive(plugin_id, version='1.2.3'):
+    author, name = plugin_id.split('/')
+    output = io.BytesIO()
+    with zipfile.ZipFile(output, 'w') as package:
+        package.writestr(
+            zipfile.ZipInfo('manifest.yaml'),
+            f'kind: Plugin\nmetadata:\n  author: {author}\n  name: {name}\n  version: {version}\n',
+        )
+        package.writestr(zipfile.ZipInfo('main.py'), '# Published fixture\n')
+    return output.getvalue()
+
+
+class PublishedPluginsTests(unittest.TestCase):
+    @classmethod
+    def setUpClass(cls):
+        class Marketplace(BaseHTTPRequestHandler):
+            def do_GET(self):
+                cls.requests.append(self.path)
+                status, body = cls.responses.get(self.path, (404, b'missing published release'))
+                self.send_response(status)
+                self.end_headers()
+                self.wfile.write(body)
+
+            def log_message(self, *args):
+                pass
+
+        cls.server = ThreadingHTTPServer(('127.0.0.1', 0), Marketplace)
+        cls.thread = threading.Thread(target=cls.server.serve_forever)
+        cls.thread.start()
+
+    @classmethod
+    def tearDownClass(cls):
+        cls.server.shutdown()
+        cls.server.server_close()
+        cls.thread.join()
+
+    def setUp(self):
+        self.folder = tempfile.TemporaryDirectory()
+        self.addCleanup(self.folder.cleanup)
+        self.root = Path(self.folder.name)
+        self.marketplace = patch.object(plugins, 'MARKETPLACE', f'http://127.0.0.1:{self.server.server_port}')
+        self.marketplace.start()
+        self.addCleanup(self.marketplace.stop)
+        type(self).requests = []
+        type(self).responses = {}
+        for plugin_id in plugins.PLUGIN_IDS:
+            author, name = plugin_id.split('/')
+            data = {
+                'code': 0,
+                'data': {
+                    'plugin': {
+                        'plugin_id': plugin_id,
+                        'author': author,
+                        'name': name,
+                        'latest_version': '1.2.3',
+                        'status': 'live',
+                    }
+                },
+            }
+            self.responses[f'/api/v1/marketplace/plugins/{plugin_id}'] = (200, json.dumps(data).encode())
+            self.responses[f'/api/v1/marketplace/plugins/download/{plugin_id}/1.2.3'] = (200, archive(plugin_id))
+
+    def test_latest_resolved_once_and_replay_uses_same_published_bytes(self):
+        directory = self.root / 'latest'
+        manifest = plugins.prepare_published_plugins(directory)
+        self.assertEqual(Counter(self.requests), {path: 1 for path in self.responses})
+        installed = []
+
+        def call(args):
+            if args[:3] == ['plugin', 'install', 'local']:
+                source = Path(args[4])
+                self.assertEqual(source.read_bytes(), archive(plugins.PLUGIN_IDS[len(installed)]))
+                author, name = plugins.PLUGIN_IDS[len(installed)].split('/')
+                installed.append(
+                    {'author': author, 'name': name, 'version': '1.2.3', 'enabled': True, 'status': 'initialized'}
+                )
+                return {}
+            self.assertEqual(args, ['plugin', 'list'])
+            return {'plugins': installed}
+
+        plugins.install_published_plugins(call, directory, manifest)
+        recorded = json.loads((directory / 'manifest.json').read_text())
+        self.assertTrue(all(entry['installed'][0]['status'] == 'initialized' for entry in recorded['plugins']))
+        self.requests.clear()
+        replay = plugins.prepare_published_plugins(self.root / 'replay', directory / 'manifest.json')
+        self.assertEqual(
+            [entry['sha256'] for entry in replay['plugins']], [entry['sha256'] for entry in manifest['plugins']]
+        )
+        self.assertEqual(len(self.requests), 3)
+        self.assertTrue(all('/download/' in path for path in self.requests))
+
+    def test_bad_metadata_and_unpublished_versions_fail_without_download(self):
+        url = '/api/v1/marketplace/plugins/langbot-team/LocalAgent'
+        valid = json.loads(self.responses[url][1])
+        invalid = [[], {'code': 0, 'data': []}, {'code': 1, 'data': valid['data']}]
+        for field, value in [
+            ('latest_version', ''),
+            ('latest_version', '../bad'),
+            ('status', 'draft'),
+            ('name', 'Other'),
+        ]:
+            data = copy.deepcopy(valid)
+            data['data']['plugin'][field] = value
+            invalid.append(data)
+        for index, payload in enumerate(invalid):
+            with self.subTest(payload=payload):
+                self.requests.clear()
+                self.responses[url] = (200, json.dumps(payload).encode())
+                with self.assertRaises(ValueError):
+                    plugins.prepare_published_plugins(self.root / str(index))
+                self.assertEqual(self.requests, [url])
+
+    def test_missing_or_wrong_release_does_not_fall_back(self):
+        url = '/api/v1/marketplace/plugins/download/langbot-team/LocalAgent/1.2.3'
+        self.responses[url] = (404, b'missing')
+        with self.assertRaisesRegex(RuntimeError, 'HTTP Error 404'):
+            plugins.prepare_published_plugins(self.root / 'missing')
+        self.responses[url] = (200, archive('langbot-team/LocalAgent', '0.0.1'))
+        with self.assertRaisesRegex(ValueError, 'identity/version differs'):
+            plugins.prepare_published_plugins(self.root / 'wrong')
+
+    def test_replay_rejects_changed_packages_and_untrusted_identity(self):
+        directory = self.root / 'latest'
+        manifest = plugins.prepare_published_plugins(directory)
+        replay = directory / 'manifest.json'
+        url = '/api/v1/marketplace/plugins/download/langbot-team/LocalAgent/1.2.3'
+        self.responses[url] = (200, b'different package')
+        with self.assertRaisesRegex(ValueError, 'sha256 changed'):
+            plugins.prepare_published_plugins(self.root / 'changed', replay)
+        for change in ('identity', 'duplicate', 'origin', 'hash'):
+            invalid = copy.deepcopy(manifest)
+            if change == 'identity':
+                invalid['plugins'][0]['id'] = 'untrusted/Other'
+            elif change == 'duplicate':
+                invalid['plugins'][1] = invalid['plugins'][0]
+            elif change == 'origin':
+                invalid['marketplace'] = 'https://untrusted.example'
+            else:
+                del invalid['plugins'][0]['sha256']
+            replay.write_text(json.dumps(invalid))
+            with self.subTest(change=change), self.assertRaises(ValueError):
+                plugins.prepare_published_plugins(self.root / change, replay)
+
+    def test_readback_records_actual_failed_version_and_stops_installing(self):
+        directory = self.root / 'latest'
+        manifest = plugins.prepare_published_plugins(directory)
+        calls = []
+
+        def call(args):
+            calls.append(args)
+            return {
+                'plugins': [
+                    {
+                        'author': 'langbot-team',
+                        'name': 'LocalAgent',
+                        'version': '0.0.1',
+                        'enabled': True,
+                        'status': 'initialized',
+                    }
+                ]
+            }
+
+        with self.assertRaisesRegex(RuntimeError, 'not ready at the selected version'):
+            plugins.install_published_plugins(call, directory, manifest)
+        self.assertEqual(len(calls), 2)
+        recorded = json.loads((directory / 'manifest.json').read_text())
+        self.assertEqual(recorded['plugins'][0]['installed'][0]['version'], '0.0.1')
+        (directory / manifest['plugins'][0]['archive']).write_bytes(b'changed after download')
+        with self.assertRaisesRegex(ValueError, 'changed before installation'):
+            plugins.install_published_plugins(call, directory, manifest)
+        self.assertEqual(len(calls), 2)
+
+
+if __name__ == '__main__':
+    unittest.main()