test(regression): add plugin lifecycle recovery job

tools/README.md

@@ -291,6 +291,35 @@ quality. Run the Job plumbing checks on Linux with:
 uv run --no-sync python -m unittest discover -s tools -p 'test_*.py' -v
 ```
 
+## Local plugin lifecycle
+
+Run the separate plugin lifecycle Job on a disposable Linux stack:
+
+```sh
+uv run --no-sync python tools/regression_ci.py --suite plugin-lifecycle \
+  --lbctl /path/to/lbctl --output .qa/plugin-lifecycle-001
+```
+
+Its three chains install a dependency-free local Page fixture, store a Unicode
+marker and configure it, then exercise an update, a repeated installation and
+invalid archive, or an initialization failure followed by explicitly reinstalling
+a known good package and updating again. Assertions check completed task status,
+the executing version, retained configuration and plugin storage, and unique
+installation visibility. Cleanup verifies API removal and termination of observed
+worker PIDs, including their process start times. The 600-second budget is a cap;
+the Job stops when its finite cases finish.
+
+The current HTTP installation path does not promise automatic rollback after
+initialization failure. This Job verifies explicit recovery; the separate
+`plugins.upgrade_recovery` coverage obligation remains unclaimed. A failed
+generation must be absent or reject Page calls. If Core instead restores the old
+version automatically, that version must pass its full configuration, storage and
+Page oracle. It uses the
+existing runner's stack preparation and local HTTP uploads, installs only the
+bundled fixtures without Marketplace access, and is not included in smoke or daily
+by default. Cases, task outcomes, worker identities and cleanup journals are saved
+under `gate/suite/`.
+
 ## Finite core smoke
 
 Use [`regression_profiles/smoke.json`](regression_profiles/smoke.json) for a finite

tools/regression_ci.py

@@ -29,6 +29,7 @@ SUITE_BUDGETS = {
     'api': 600,
     'state': 300,
     'isolation': 300,
+    'plugin-lifecycle': 600,
 }
 sys.path.insert(0, str(ROOT))
 from tests.e2e.utils.config_factory import create_minimal_config, create_test_directories  # noqa: E402
@@ -56,6 +57,11 @@ def suite_profile(name, run_id, groups=(), api_seed=1):
         from regression_isolation_cases import isolation_profile
 
         profile = isolation_profile(profile)
+    if name == 'plugin-lifecycle':
+        from regression_plugin_lifecycle import add_cases
+
+        profile.update(resources=[], cases=[], coverage_suite='plugin-lifecycle')
+        add_cases(profile)
     for resource in profile['resources']:
         if resource['kind'] == 'pipeline':
             resource['body'].setdefault(
@@ -177,11 +183,15 @@ def run(args):
         return {'service': name, 'old_pid': old_pid, 'new_pid': pids[name], 'old_exit_code': process.returncode}
 
     try:
-        packages = (
-            prepare_source_plugins(root / 'plugins', args.plugins, args.extra_plugins)
-            if args.plugins is not None
-            else prepare_published_plugins(root / 'plugins', args.plugin_manifest)
-        )
+        if args.suite == 'plugin-lifecycle':
+            (root / 'plugins').mkdir()
+            packages = {'version': 1, 'selection': 'local-fixtures', 'plugins': []}
+        else:
+            packages = (
+                prepare_source_plugins(root / 'plugins', args.plugins, args.extra_plugins)
+                if args.plugins is not None
+                else 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())
@@ -195,7 +205,7 @@ def run(args):
         groups = args.scenario_groups
         if groups is None:
             groups = list(DEFAULT_GROUPS) if args.suite in ('scenarios', 'daily') else []
-            if args.suite in ('api', 'isolation'):
+            if args.suite in ('api', 'isolation', 'plugin-lifecycle'):
                 groups = [args.suite]
             if 'plugins' in groups and not {'misc/EssentialCommands', 'misc/FAQManager'} <= set(args.extra_plugins):
                 groups.remove('plugins')

tools/regression_contracts.json

@@ -256,6 +256,28 @@
      "description": "Plugin upgrade failure preserves a usable installation",
      "bindings": []
     },
+    {"id": "plugins.upgrade_persistence",
+     "area": "plugins", "kind": "blackbox", "priority": "P1",
+     "description": "A local plugin update runs the new version with its existing config and stored data",
+     "bindings": [{"profile": "plugin-lifecycle", "case": "plugin_upgrade", "checks": ["upgrade_v2_task", "v2_identity", "v2_unique", "v2_state", "plugin_cleanup"]}],
+     "routes": ["POST /api/v1/plugins/install/local", "GET /api/v1/plugins/<author>/<plugin_name>", "PUT /api/v1/plugins/<author>/<plugin_name>/config", "POST /api/v1/plugins/<author>/<plugin_name>/page-api"]
+    },
+    {"id": "plugins.package_rejection",
+     "area": "plugins", "kind": "blackbox", "priority": "P1",
+     "description": "Reinstalling a local package preserves state and an invalid archive fails without disrupting that installation",
+     "bindings": [{"profile": "plugin-lifecycle", "case": "plugin_rejected_package", "checks": ["reinstall_v1_task", "after_reinstall_state", "invalid_archive_task", "after_invalid_archive_identity", "after_invalid_archive_state", "plugin_cleanup"]}]
+    },
+    {"id": "plugins.explicit_recovery",
+     "area": "plugins", "kind": "blackbox", "priority": "P1",
+     "description": "A failed plugin initialization reports failure and reinstalling a known good package restores persisted state and later upgrades",
+     "bindings": [{"profile": "plugin-lifecycle", "case": "plugin_failed_activation", "checks": ["broken_v2_task", "broken_v2_error", "failed_generation_safe", "restore_v1_task", "restored_v1_state", "recovered_v3_task", "v3_state", "plugin_cleanup"]}]
+    },
+    {"id": "plugins.observed_worker_cleanup",
+     "area": "plugins", "kind": "graybox", "priority": "P1",
+     "description": "Uninstall removes each fixture from the API and stops its observed worker process identities",
+     "bindings": [{"profile": "plugin-lifecycle", "case": "plugin_upgrade", "checks": ["uninstall_task", "uninstalled_from_list", "uninstalled_workers_stopped"]}, {"profile": "plugin-lifecycle", "case": "plugin_rejected_package", "checks": ["uninstall_task", "uninstalled_from_list", "uninstalled_workers_stopped"]}, {"profile": "plugin-lifecycle", "case": "plugin_failed_activation", "checks": ["uninstall_task", "uninstalled_from_list", "uninstalled_workers_stopped"]}],
+     "routes": ["DELETE /api/v1/plugins/<author>/<plugin_name>"]
+    },
     {"id": "agent.direct_execution",
      "area": "agent", "kind": "blackbox", "priority": "P1",
      "description": "Standalone Agent entrypoint completes through the public API",

tools/regression_fixtures/plugin_lifecycle/components/pages/probe/index.html

@@ -0,0 +1 @@
+<!doctype html><title>Lifecycle probe</title>

tools/regression_fixtures/plugin_lifecycle/components/pages/probe/probe.py

@@ -0,0 +1,20 @@
+import os
+
+from langbot_plugin.api.definition.components.page import Page, PageRequest, PageResponse
+
+
+class ProbePage(Page):
+    async def handle_api(self, request: PageRequest) -> PageResponse:
+        if request.endpoint != '/state' or request.method not in ('GET', 'POST'):
+            return PageResponse.fail('Unsupported lifecycle probe request')
+        if request.method == 'POST':
+            await self.plugin.set_plugin_storage('probe', request.body['token'].encode())
+        token = (await self.plugin.get_plugin_storage('probe')).decode()
+        return PageResponse.ok(
+            {
+                'version': self.plugin.version,
+                'token': token,
+                'config': self.plugin.get_config(),
+                'pid': os.getpid(),
+            }
+        )

tools/regression_fixtures/plugin_lifecycle/components/pages/probe/probe.yaml

@@ -0,0 +1,12 @@
+apiVersion: v1
+kind: Page
+metadata:
+  name: probe
+  label:
+    en_US: Lifecycle probe
+spec:
+  path: index.html
+execution:
+  python:
+    path: probe.py
+    attr: ProbePage

tools/regression_fixtures/plugin_lifecycle/main.py

@@ -0,0 +1,12 @@
+from langbot_plugin.api.definition.plugin import BasePlugin
+
+VERSION = '0.0.0'
+FAIL_INITIALIZE = False
+
+
+class LifecyclePlugin(BasePlugin):
+    version = VERSION
+
+    async def initialize(self):
+        if FAIL_INITIALIZE:
+            raise RuntimeError('QA deliberate initialization failure')

tools/regression_local_cases.py

@@ -16,6 +16,7 @@ GROUPS = {
     'plugins': 'regression_plugin_cases',
     'api': 'regression_api_cases',
     'isolation': 'regression_isolation_cases',
+    'plugin-lifecycle': 'regression_plugin_lifecycle',
 }
 
 

tools/regression_plugin_lifecycle.py

@@ -0,0 +1,285 @@
+"""Exercise local plugin generations through HTTP tasks and real Page invocations."""
+
+import io
+import json
+from pathlib import Path
+import time
+import urllib.request
+from uuid import uuid4
+import zipfile
+
+import yaml
+
+from regression_api_checks import check, request
+from regression_coverage import checked
+from regression_local_cases import api, auth_headers, endpoint, require
+from regression_resource_cases import cleanup, journal
+
+CASE_NAMES = ('plugin_upgrade', 'plugin_rejected_package', 'plugin_failed_activation')
+FIXTURE = Path(__file__).with_name('regression_fixtures') / 'plugin_lifecycle'
+AUTHOR = 'regression-qa'
+
+
+def add_cases(profile):
+    require(profile.get('coverage_suite') == 'plugin-lifecycle', 'Plugin lifecycle requires its dedicated suite')
+    profile['scenario_groups'] = ['plugin-lifecycle']
+    profile['cases'].extend(
+        {'name': name, 'scenario': name, 'stage': 'plugins', 'expect': 'PLUGIN_LIFECYCLE_OK'} for name in CASE_NAMES
+    )
+
+
+def package(name, version, *, broken=False):
+    manifest = {
+        'apiVersion': 'v1',
+        'kind': 'Plugin',
+        'metadata': {
+            'author': AUTHOR,
+            'name': name,
+            'version': version,
+            'label': {'en_US': 'Lifecycle probe'},
+            'description': {'en_US': 'Local regression fixture'},
+            'icon': 'icon.svg',
+        },
+        'spec': {
+            'config': [{'name': 'marker', 'type': 'string', 'default': '', 'label': {'en_US': 'Marker'}}],
+            'components': {'Page': {'fromDirs': [{'path': 'components/pages/', 'maxDepth': 2}]}},
+        },
+        'execution': {'python': {'path': 'main.py', 'attr': 'LifecyclePlugin'}},
+    }
+    buffer = io.BytesIO()
+    with zipfile.ZipFile(buffer, 'w', zipfile.ZIP_DEFLATED) as archive:
+        archive.writestr('manifest.yaml', yaml.safe_dump(manifest))
+        archive.writestr('icon.svg', '<svg xmlns="http://www.w3.org/2000/svg"/>')
+        for path in sorted(FIXTURE.rglob('*')):
+            if not path.is_file() or '__pycache__' in path.parts:
+                continue
+            content = path.read_text()
+            if path.name == 'main.py':
+                content = content.replace("VERSION = '0.0.0'", f'VERSION = {version!r}')
+                if broken:
+                    content = content.replace('FAIL_INITIALIZE = False', 'FAIL_INITIALIZE = True')
+            archive.writestr(str(path.relative_to(FIXTURE)), content)
+    return buffer.getvalue()
+
+
+def wait_task(suite, row, task_id, expected, label):
+    deadline = time.monotonic() + suite.timeout(90)
+    require(isinstance(task_id, int), 'Plugin operation did not return a task ID')
+    while True:
+        task = request(suite, row, 'GET', f'/api/v1/system/tasks/{task_id}')
+        require(task.get('id') == task_id, 'Plugin task identity changed')
+        status = task.get('status')
+        require(status in {'running', 'succeeded', 'failed', 'cancelled'}, 'Malformed plugin task status')
+        if status != 'running':
+            row.setdefault('plugin_tasks', []).append({'label': label, **task})
+            check(row, label + '_task', status == expected, f'{label}: expected {expected}, got {status}')
+            if expected == 'failed':
+                check(row, label + '_error', bool(task.get('error')), 'Failed task omitted its error')
+            return
+        if time.monotonic() >= deadline:
+            raise TimeoutError(f'{label}: plugin task did not finish')
+        time.sleep(0.2)
+
+
+def upload(suite, row, content, name, *, expected='succeeded'):
+    boundary = 'qa' + uuid4().hex
+    body = (
+        (
+            f'--{boundary}\r\nContent-Disposition: form-data; name="file"; filename="{name}.lbpkg"\r\n'
+            'Content-Type: application/octet-stream\r\n\r\n'
+        ).encode()
+        + content
+        + f'\r\n--{boundary}--\r\n'.encode()
+    )
+    headers = auth_headers(suite)
+    headers['Content-Type'] = 'multipart/form-data; boundary=' + boundary
+    path = '/api/v1/plugins/install/local'
+    call = urllib.request.Request(endpoint(suite, path), data=body, headers=headers, method='POST')
+    opener = urllib.request.build_opener(urllib.request.ProxyHandler({}))
+    with opener.open(call, timeout=suite.timeout(90)) as response:
+        result = json.load(response)
+        check(
+            row,
+            name + '_upload',
+            response.status == 200 and result.get('code') == 0,
+            'Local plugin upload was not accepted',
+        )
+    wait_task(suite, row, result['data']['task_id'], expected, name)
+
+
+def process_identity(pid):
+    try:
+        # Start time distinguishes an old worker from a reused PID.
+        return Path(f'/proc/{pid}/stat').read_text().rsplit(')', 1)[1].split()[19]
+    except FileNotFoundError:
+        return None
+
+
+def metadata(plugin):
+    return plugin['manifest']['manifest']['metadata']
+
+
+def failed_generation_safe(suite, row, path, verify_good):
+    def verify():
+        status, response = api(suite, 'GET', path)
+        row['failed_generation'] = {'http_status': status, 'plugin': response.get('data', {}).get('plugin')}
+        if status == 404:
+            require(response.get('code') not in (None, 0), 'Missing plugin returned a successful business status')
+            return
+        require(status == 200 and response.get('code') == 0, 'Failed generation cannot be observed')
+        info = response.get('data', {}).get('plugin')
+        require(isinstance(info, dict), 'Failed generation metadata is missing')
+        identity = metadata(info)
+        require(
+            identity.get('author') == AUTHOR and identity.get('name') == path.rsplit('/', 1)[1],
+            'Failed generation identity changed',
+        )
+        if identity.get('version') == '0.0.1':
+            verify_good()  # A future automatic restore must pass the complete old-version state oracle.
+            return
+        require(
+            identity.get('version') == '0.0.2' and info.get('status') in {'unmounted', 'mounted'},
+            'Failed plugin generation is initialized or has an unexpected version/state',
+        )
+        page_status, page = api(
+            suite,
+            'POST',
+            path + '/page-api',
+            {
+                'page_id': 'probe',
+                'endpoint': '/state',
+                'method': 'GET',
+                'body': None,
+            },
+        )
+        row['failed_generation']['page_status'] = page_status
+        require(
+            page_status in {400, 404, 409, 500, 503} and page.get('code') not in (None, 0),
+            'Failed plugin generation remains callable',
+        )
+
+    checked(row, 'failed_generation_safe', verify)
+
+
+def workers_stopped(suite, workers):
+    deadline = time.monotonic() + suite.timeout(10)
+    while any(process_identity(int(pid)) == started for pid, started in workers.items()):
+        if time.monotonic() >= deadline:
+            raise ValueError('An observed plugin worker remains after uninstall or upgrade')
+        time.sleep(0.1)
+
+
+def run_case(suite, case, row):
+    name = 'lifecycle-' + uuid4().hex
+    path = f'/api/v1/plugins/{AUTHOR}/{name}'
+    owned = {'kind': 'plugin', 'author': AUTHOR, 'name': name, 'state': 'creating'}
+    row.setdefault('owned_resources', []).append(owned)
+    journal(suite, row)
+    workers = row['plugin_workers'] = {}
+    marker = 'persistent-中文-' + uuid4().hex
+
+    def installed(version, label, *, write=False):
+        deadline = time.monotonic() + suite.timeout(45)
+        while True:
+            status, response = api(suite, 'GET', path)
+            info = response.get('data', {}).get('plugin', {})
+            if status == 200 and info.get('status') == 'initialized':
+                break
+            require(status in {200, 404}, 'Plugin readiness query failed')
+            if time.monotonic() >= deadline:
+                raise TimeoutError('Plugin did not become ready after operation')
+            time.sleep(0.2)
+        identity = metadata(info)
+        check(
+            row,
+            label + '_identity',
+            identity.get('author') == AUTHOR
+            and identity.get('name') == name
+            and identity.get('version') == version
+            and info.get('enabled') is True,
+            'Plugin generation differs',
+        )
+        listed = request(suite, row, 'GET', '/api/v1/plugins')['plugins']
+        check(
+            row,
+            label + '_unique',
+            sum(metadata(p)['author'] == AUTHOR and metadata(p)['name'] == name for p in listed) == 1,
+            'Plugin generation is missing or duplicated',
+        )
+        state = request(
+            suite,
+            row,
+            'POST',
+            path + '/page-api',
+            {
+                'page_id': 'probe',
+                'endpoint': '/state',
+                'method': 'POST' if write else 'GET',
+                'body': {'token': marker} if write else None,
+            },
+        )
+        check(
+            row,
+            label + '_state',
+            state.get('version') == version
+            and state.get('token') == marker
+            and state.get('config') == {'marker': marker},
+            'Plugin call lost version, stored data, or config',
+        )
+        pid = state.get('pid')
+        require(isinstance(pid, int) and pid > 0, 'Plugin did not return its worker PID')
+        started = process_identity(pid)
+        require(started is not None, 'Observed plugin worker already disappeared')
+        old = {key: value for key, value in workers.items() if key != str(pid)}
+        workers[str(pid)] = started
+        checked(row, label + '_old_workers_stopped', lambda: workers_stopped(suite, old))
+        return state
+
+    def remove():
+        status, response = api(suite, 'GET', path)
+        if status != 404:
+            require(status == 200 and response.get('code') == 0, 'Cannot confirm owned plugin before cleanup')
+            info = metadata(response['data']['plugin'])
+            require(info.get('author') == AUTHOR and info.get('name') == name, 'Plugin cleanup identity changed')
+        # A failed generation can have a durable installation without a registered worker.
+        task = request(suite, row, 'DELETE', path + '?delete_data=true')
+        wait_task(suite, row, task['task_id'], 'succeeded', 'uninstall')
+        request(suite, row, 'GET', path, expected=404)
+        listed = request(suite, row, 'GET', '/api/v1/plugins')['plugins']
+        check(
+            row,
+            'uninstalled_from_list',
+            not any(metadata(p)['author'] == AUTHOR and metadata(p)['name'] == name for p in listed),
+            'Uninstalled plugin remains listed',
+        )
+        checked(row, 'uninstalled_workers_stopped', lambda: workers_stopped(suite, workers))
+        owned['state'] = 'deleted'
+
+    request(suite, row, 'GET', path, expected=404)
+    try:
+        first = package(name, '0.0.1')
+        upload(suite, row, first, 'install_v1')
+        owned['state'] = 'installed'
+        journal(suite, row)
+        request(suite, row, 'PUT', path + '/config', {'marker': marker})
+        installed('0.0.1', 'v1', write=True)
+        if case['scenario'] == 'plugin_upgrade':
+            upload(suite, row, package(name, '0.0.2'), 'upgrade_v2')
+            installed('0.0.2', 'v2')
+        elif case['scenario'] == 'plugin_rejected_package':
+            upload(suite, row, first, 'reinstall_v1')
+            installed('0.0.1', 'after_reinstall')
+            upload(suite, row, b'not a plugin zip', 'invalid_archive', expected='failed')
+            installed('0.0.1', 'after_invalid_archive')
+        elif case['scenario'] == 'plugin_failed_activation':
+            upload(suite, row, package(name, '0.0.2', broken=True), 'broken_v2', expected='failed')
+            failed_generation_safe(suite, row, path, lambda: installed('0.0.1', 'automatic_restore_v1'))
+            upload(suite, row, first, 'restore_v1')
+            installed('0.0.1', 'restored_v1')
+            upload(suite, row, package(name, '0.0.3'), 'recovered_v3')
+            installed('0.0.3', 'v3')
+        else:
+            raise ValueError('Unknown plugin lifecycle case')
+        return {'result': case['expect']}
+    finally:
+        cleanup(suite, row, 'plugin_cleanup', remove)

tools/test_regression_coverage.py

@@ -289,7 +289,7 @@ class Routes:
                 name = binding['profile']
                 if name not in profiles:
                     path = Path(__file__).with_name('regression_profiles') / f'{name}.json'
-                    if name in ('api', 'state', 'isolation'):
+                    if name in ('api', 'state', 'isolation', 'plugin-lifecycle'):
                         from regression_ci import suite_profile
 
                         profile, _ = suite_profile(name, 'catalog-test')

tools/test_regression_plugin_lifecycle.py

@@ -0,0 +1,118 @@
+"""Challenge plugin lifecycle task verdicts and worker cleanup evidence."""
+
+from types import SimpleNamespace
+import unittest
+from unittest.mock import Mock, patch
+
+from regression_ci import suite_profile
+from regression_local_cases import DEFAULT_GROUPS
+from regression_plugin_lifecycle import failed_generation_safe, wait_task, workers_stopped
+from regression_suite import validate
+
+
+class PluginLifecycleTests(unittest.TestCase):
+    def test_failed_generation_cannot_pass_by_http_status_alone(self):
+        path = '/api/v1/plugins/regression-qa/probe'
+
+        def plugin(version, status):
+            return (
+                200,
+                {
+                    'code': 0,
+                    'data': {
+                        'plugin': {
+                            'status': status,
+                            'manifest': {
+                                'manifest': {
+                                    'metadata': {'author': 'regression-qa', 'name': 'probe', 'version': version},
+                                }
+                            },
+                        }
+                    },
+                },
+            )
+
+        for responses in (
+            [(404, {'code': 0})],
+            [plugin('0.0.2', 'initialized')],
+            [plugin('0.0.9', 'mounted')],
+            [plugin('0.0.2', 'mounted'), (200, {'code': 0, 'data': {}})],
+            [plugin('0.0.2', 'mounted'), (500, {'code': 0})],
+        ):
+            row = {}
+            with self.subTest(responses=responses), patch('regression_plugin_lifecycle.api', side_effect=responses):
+                with self.assertRaises(ValueError):
+                    failed_generation_safe(None, row, path, Mock())
+                self.assertEqual(row['assertions']['failed_generation_safe'], 'failed')
+
+        for responses in (
+            [(404, {'code': -1})],
+            [plugin('0.0.2', 'mounted'), (400, {'code': -1})],
+        ):
+            row, verify_good = {}, Mock()
+            with self.subTest(responses=responses), patch('regression_plugin_lifecycle.api', side_effect=responses):
+                failed_generation_safe(None, row, path, verify_good)
+                self.assertEqual(row['assertions']['failed_generation_safe'], 'passed')
+                verify_good.assert_not_called()
+
+        for state_error in (None, ValueError('Restored plugin lost configuration or stored data')):
+            row, verify_good = {}, Mock(side_effect=state_error)
+            with (
+                self.subTest(state_error=state_error),
+                patch('regression_plugin_lifecycle.api', return_value=plugin('0.0.1', 'initialized')),
+            ):
+                if state_error:
+                    with self.assertRaisesRegex(ValueError, 'lost configuration'):
+                        failed_generation_safe(None, row, path, verify_good)
+                    self.assertEqual(row['assertions']['failed_generation_safe'], 'failed')
+                else:
+                    failed_generation_safe(None, row, path, verify_good)
+                    self.assertEqual(row['assertions']['failed_generation_safe'], 'passed')
+                verify_good.assert_called_once_with()
+
+    def test_job_is_opt_in_and_does_not_expand_daily(self):
+        profile, _ = suite_profile('plugin-lifecycle', 'run')
+        validate(profile)
+        self.assertEqual(len(profile['cases']), 3)
+        self.assertEqual(profile['resources'], [])
+        self.assertNotIn('plugin-lifecycle', DEFAULT_GROUPS)
+        for name in ('smoke', 'daily'):
+            existing, _ = suite_profile(name, 'run')
+            self.assertFalse(any(c.get('scenario', '').startswith('plugin_') for c in existing['cases']))
+
+    def test_expected_failure_requires_a_failed_task_not_http_success(self):
+        suite = SimpleNamespace(timeout=lambda seconds: seconds)
+        for task in (
+            {'id': 7, 'status': 'succeeded'},
+            {'id': 7, 'status': 'cancelled'},
+            {'id': 7, 'status': 'failed', 'error': None},
+            {'id': 8, 'status': 'failed', 'error': {'type': 'task_failed'}},
+        ):
+            with self.subTest(task=task), patch('regression_plugin_lifecycle.request', return_value=task):
+                with self.assertRaises(ValueError):
+                    wait_task(suite, {}, 7, 'failed', 'broken_upgrade')
+        row = {}
+        task = {'id': 7, 'status': 'failed', 'error': {'type': 'task_failed'}}
+        with patch('regression_plugin_lifecycle.request', return_value=task):
+            wait_task(suite, row, 7, 'failed', 'broken_upgrade')
+        self.assertEqual(row['assertions']['broken_upgrade_task'], 'passed')
+        self.assertEqual(row['plugin_tasks'][0]['status'], 'failed')
+
+    def test_live_worker_is_a_cleanup_failure_but_reused_pid_is_not(self):
+        suite = SimpleNamespace(timeout=lambda seconds: 0)
+        with patch('regression_plugin_lifecycle.process_identity', return_value='same-start'):
+            with self.assertRaisesRegex(ValueError, 'worker remains'):
+                workers_stopped(suite, {'100': 'same-start'})
+        for identity in (None, 'new-start'):
+            with patch('regression_plugin_lifecycle.process_identity', return_value=identity):
+                workers_stopped(suite, {'100': 'same-start'})
+
+    def test_unfinished_task_cannot_be_accepted(self):
+        suite = SimpleNamespace(timeout=lambda seconds: 0)
+        with patch('regression_plugin_lifecycle.request', Mock(return_value={'id': 7, 'status': 'running'})):
+            with self.assertRaisesRegex(TimeoutError, 'did not finish'):
+                wait_task(suite, {}, 7, 'succeeded', 'install')
+
+
+if __name__ == '__main__':
+    unittest.main()