diff --git a/hydra/build.py b/hydra/build.py
index 7254f3a..661d5b2 100644
--- a/hydra/build.py
+++ b/hydra/build.py
@@ -4,8 +4,10 @@ import contextlib
 import fcntl
 import hashlib
 import json
+import lzma
 import os
 from pathlib import Path
+import re
 import shutil
 import subprocess
 import sys
@@ -18,6 +20,43 @@ class BuildError(RuntimeError):
     pass
 
 
+def package_job_script(command, state='/var/lib/hydra-build', package_db='/var/db/pkg', inputs=''):
+    """The contained job owns its log/result even if the outer caller disconnects."""
+    return f'''import fcntl, json, os, pathlib, subprocess, sys, time
+state = pathlib.Path({str(state)!r})
+database = pathlib.Path({str(package_db)!r})
+state.mkdir(parents=True, exist_ok=True)
+lock = (state / 'packages.lock').open('a')
+try:
+    fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB)
+except BlockingIOError:
+    sys.exit(75)
+def write(name, data):
+    temporary = state / (name + '.tmp')
+    temporary.write_text(json.dumps(data, indent=2) + '\\n')
+    temporary.replace(state / name)
+status = {{'running': True, 'pid': os.getpid(), 'inputs': {inputs!r}, 'started': int(time.time())}}
+write('packages.status.json', status)
+with (state / 'packages.log').open('a') as log:
+    try:
+        result = subprocess.run({command!r}, stdout=log, stderr=subprocess.STDOUT)
+        code = result.returncode
+    except OSError as error:
+        log.write(str(error) + '\\n')
+        code = 127
+if code == 0:
+    inventory = []
+    for use in sorted(database.glob('*/*/USE')):
+        package = use.parent
+        inventory.append({{'package': str(package.relative_to(database)), 'use': use.read_text().split(),
+                          'build_time': (package / 'BUILD_TIME').read_text().strip() if (package / 'BUILD_TIME').exists() else None}})
+    write('packages.json', inventory)
+status.update(running=False, exit_code=code, finished=int(time.time()))
+write('packages.status.json', status)
+sys.exit(code)
+'''
+
+
 def verify_hash(path, expected):
     digest = hashlib.sha256()
     with Path(path).open('rb') as stream:
@@ -54,6 +93,7 @@ class Builder:
         self.runroot = runtime / ('hydra-build-podman' if self.work == self.project / 'build' else f'hydra-build-{identifier}')
         self.name = 'hydra-builder-' + identifier
         self.image = 'localhost/hydra-stage3:20261011'
+        self._stage3_layers = {}
 
     def podman_argv(self, *args):
         return [self.podman, '--root', str(self.work / 'podman'), '--runroot', str(self.runroot), '--storage-driver', 'overlay', *args]
@@ -65,7 +105,7 @@ class Builder:
                 self.image, '/bin/sleep', 'infinity']
 
     def run(self, argv, capture=False, check=True):
-        print('+ ' + ' '.join(str(arg) for arg in argv), flush=True)
+        print('+ ' + ' '.join(str(arg) for arg in argv), file=sys.stderr, flush=True)
         result = subprocess.run(argv, text=True, stdout=subprocess.PIPE if capture else None,
                                 stderr=subprocess.PIPE if capture else None)
         if check and result.returncode:
@@ -78,6 +118,72 @@ class Builder:
     def exec(self, command, capture=False):
         return self.pod('exec', self.name, '/bin/bash', '-euc', command, capture=capture)
 
+    def inspect(self, kind, name):
+        result = self.pod(kind, 'inspect', name, capture=True, check=False)
+        if result.returncode:
+            raise BuildError(f'{kind} provenance is unavailable; use a fresh work directory')
+        entries = json.loads(result.stdout)
+        if not isinstance(entries, list) or len(entries) != 1 or not isinstance(entries[0], dict):
+            raise BuildError(f'Unexpected {kind} provenance')
+        return entries[0]
+
+    def check_mounts(self, container):
+        for destination, path, label in (('/hydra-source', self.source_dir, 'source'),
+                                         ('/hydra-cache', self.cache, 'cache')):
+            mounts = [item for item in container.get('Mounts', []) if item.get('Destination') == destination]
+            if len(mounts) != 1 or mounts[0].get('Source') != str(path) or mounts[0].get('RW') is not False:
+                raise BuildError(f'Existing builder {label} mount differs; use its original checkout/cache or a fresh work directory')
+
+    def stage3_layer(self, inputs):
+        stage = inputs['stage3']
+        archive = self.cache / Path(urllib.parse.urlparse(stage['url']).path).name
+        verify_hash(archive, stage['sha256'])
+        if stage['sha256'] not in self._stage3_layers:
+            digest = hashlib.sha256()
+            with lzma.open(archive, 'rb') as stream:
+                for chunk in iter(lambda: stream.read(1024 * 1024), b''):
+                    digest.update(chunk)
+            self._stage3_layers[stage['sha256']] = 'sha256:' + digest.hexdigest()
+        return self._stage3_layers[stage['sha256']]
+
+    def check_base(self, image, inputs):
+        expected = self.stage3_layer(inputs)
+        if image.get('RootFS', {}).get('Layers') != [expected]:
+            raise BuildError('Existing builder stage3 differs from pinned archive; use a fresh work directory')
+        return expected
+
+    def read_container_file(self, path):
+        # Podman cp also reads a stopped container, so provenance precedes start.
+        with tempfile.TemporaryDirectory(prefix='hydra-read-') as directory:
+            target = Path(directory) / 'data'
+            result = self.pod('cp', f'{self.name}:{path}', str(target), capture=True, check=False)
+            return target.read_text() if result.returncode == 0 else None
+
+    def check_snapshot(self, inputs, allow_pending=False):
+        marker = self.read_container_file('/var/db/repos/gentoo/.hydra-snapshot')
+        if marker is None and allow_pending:
+            marker = self.read_container_file('/var/db/repos/gentoo/.hydra-snapshot-input')
+        if marker is None or marker.strip() != inputs['portage']['commit']:
+            raise BuildError('Existing builder snapshot differs or is unverified; use a fresh work directory')
+
+    def active_package_job(self):
+        if not (self.work / 'podman').is_dir():
+            return False
+        result = self.pod('top', self.name, 'hpid', 'args', capture=True, check=False)
+        if result.returncode == 0 and any(re.search(r'(?:^|\s)(?:\S*/)?emerge(?:\s|$)', line)
+                                         for line in result.stdout.splitlines()[1:]):
+            return True
+        probe = self.pod('exec', self.name, 'python3', '-c', '''import fcntl, pathlib, sys
+path = pathlib.Path('/var/lib/hydra-build/packages.lock')
+if path.exists():
+    with path.open('r') as lock:
+        try:
+            fcntl.flock(lock, fcntl.LOCK_SH | fcntl.LOCK_NB)
+        except BlockingIOError:
+            sys.exit(75)
+''', capture=True, check=False)
+        return probe.returncode == 75
+
     def manifest(self):
         path = self.work / 'manifest.json'
         return json.loads(path.read_text()) if path.exists() else {}
@@ -106,9 +212,17 @@ class Builder:
         if step == 'packages':
             if not (self.work / 'artifacts/packages.json').is_file():
                 return False
-            return self.pod('container', 'exists', self.name, capture=True, check=False).returncode == 0
+            return self.complete('prepare')
         if step == 'prepare':
-            return self.pod('container', 'exists', self.name, capture=True, check=False).returncode == 0
+            try:
+                container = self.inspect('container', self.name)
+                self.check_mounts(container)
+                inputs = json.loads((self.source_dir / 'image/sources.json').read_text())
+                self.check_base(self.inspect('image', container['Image']), inputs)
+                self.check_snapshot(inputs)
+                return True
+            except (BuildError, OSError, ValueError, KeyError, lzma.LZMAError):
+                return False
         if step == 'fetch':
             inputs = json.loads((self.source_dir / 'image/sources.json').read_text())
             try:
@@ -153,10 +267,22 @@ class Builder:
         return inputs
 
     def prepare(self):
+        existing = self.pod('container', 'exists', self.name, capture=True, check=False).returncode == 0
+        container = self.inspect('container', self.name) if existing else None
+        if container is not None:
+            self.check_mounts(container)  # Reject another worktree before changing any checkpoint.
+            if self.active_package_job():
+                raise BuildError('A contained package job is running; wait for it to finish before preparation')
         inputs = self.fetch()  # Reverify cached bytes even when resuming.
-        if self.pod('image', 'exists', self.image, capture=True, check=False).returncode:
+        if container is not None:
+            layer = self.check_base(self.inspect('image', container['Image']), inputs)
+            self.check_snapshot(inputs, allow_pending=True)
+        elif self.pod('image', 'exists', self.image, capture=True, check=False).returncode:
             self.pod('import', str(self.cache / Path(urllib.parse.urlparse(inputs['stage3']['url']).path).name), self.image)
-        if self.pod('container', 'exists', self.name, capture=True, check=False).returncode:
+            layer = self.check_base(self.inspect('image', self.image), inputs)
+        else:
+            layer = self.check_base(self.inspect('image', self.image), inputs)
+        if container is None:
             self.pod(*self.container_run_args())
         else:
             self.pod('start', self.name)
@@ -164,6 +290,7 @@ class Builder:
         # Snapshot extraction happens only inside the container; host paths are read-only.
         self.exec(f'''mkdir -p /var/db/repos/gentoo
 if [ ! -f /var/db/repos/gentoo/.hydra-snapshot ]; then
+  printf '%s\\n' '{inputs['portage']['commit']}' > /var/db/repos/gentoo/.hydra-snapshot-input
   tar --extract --gzip --file /hydra-cache/{snapshot} --strip-components=1 --directory /var/db/repos/gentoo
   printf '%s\\n' '{inputs['portage']['commit']}' > /var/db/repos/gentoo/.hydra-snapshot
 fi
@@ -175,23 +302,50 @@ getuto
 gpg --batch --homedir /etc/portage/gnupg --import /hydra-cache/service-keys.gpg
 chmod a+r /etc/portage/gnupg/pubring.kbx /etc/portage/gnupg/trustdb.gpg
 ''')
-        self.record('prepare', image=self.image, snapshot=inputs['portage']['commit'])
+        self.record('prepare', image=container['Image'] if container else self.image,
+                    stage3_sha256=inputs['stage3']['sha256'], stage3_layer=layer,
+                    source_dir=str(self.source_dir), cache_dir=str(self.cache), snapshot=inputs['portage']['commit'])
 
     def packages(self, pretend=False):
         self.prepare()
         atoms = [line.strip() for line in (self.source_dir / 'image/packages').read_text().splitlines()
                  if line.strip() and not line.startswith('#')]
         import shlex
-        command = 'emerge --verbose --update --newuse --deep --backtrack=100 ' + ('--pretend ' if pretend else '') + ' '.join(map(shlex.quote, atoms))
-        self.exec(command)
+        command = ['emerge', '--verbose', '--update', '--newuse', '--deep', '--backtrack=100']
         if pretend:
+            self.exec(shlex.join([*command, '--pretend', *atoms]))
             return
-        inventory = self.exec('python3 - <<\'PY\'\nimport json,pathlib\nout=[]\nfor use in sorted(pathlib.Path("/var/db/pkg").glob("*/*/USE")):\n p=use.parent\n out.append({"package":str(p.relative_to("/var/db/pkg")),"use":use.read_text().split(),"build_time":(p/"BUILD_TIME").read_text().strip() if (p/"BUILD_TIME").exists() else None})\nprint(json.dumps(out,indent=2))\nPY', capture=True).stdout
+        inputs = self.input_id()
+        self.record('packages', complete=False, job_log='/var/lib/hydra-build/packages.log')
+        script = package_job_script([*command, *atoms], inputs=inputs)
+        self.exec(shlex.join(['python3', '-c', script]))
+        if self.input_id() != inputs:
+            raise BuildError('Image inputs changed during package installation; completion was not recorded')
+        inventory = self.exec('cat /var/lib/hydra-build/packages.json', capture=True).stdout
         (self.work / 'artifacts/packages.json').write_text(inventory)
         self.record('packages', inventory='artifacts/packages.json')
 
     def status(self):
-        print(json.dumps(self.manifest(), indent=2))
+        data = self.manifest()
+        for step in ('fetch', 'prepare', 'packages', 'rootfs', 'iso'):
+            if step in data:
+                data[step]['recorded_complete'] = data[step].get('complete', False)
+                data[step]['complete'] = self.complete(step)
+        data['mutation_lock_held'] = self.build_running()
+        data['package_job_running'] = self.active_package_job()
+        data['running'] = data['mutation_lock_held'] or data['package_job_running']
+        print(json.dumps(data, indent=2))
+
+    def build_running(self):
+        try:
+            with (self.work / '.hydra-build.lock').open('r') as lock:
+                try:
+                    fcntl.flock(lock, fcntl.LOCK_SH | fcntl.LOCK_NB)
+                except BlockingIOError:
+                    return True
+        except FileNotFoundError:
+            pass
+        return False
 
     @contextlib.contextmanager
     def locked(self):
@@ -222,6 +376,9 @@ def main(argv=None):
     options = parser.parse_args(argv)
     try:
         builder = Builder(project, options.work, options.cache, options.podman, options.gpg, source)
+        if options.step == 'status':
+            builder.status()
+            return 0
         with builder.locked():
             if options.step in ('rootfs', 'iso'):
                 raise BuildError(f'{options.step} assembly is not configured yet')
@@ -230,6 +387,6 @@ def main(argv=None):
             else:
                 getattr(builder, options.step)()
         return 0
-    except (BuildError, OSError, ValueError) as exc:
+    except (BuildError, OSError, ValueError, lzma.LZMAError) as exc:
         print(f'hydra-build: {exc}', file=sys.stderr)
         return 1
diff --git a/tests/test_build.py b/tests/test_build.py
index d0ccb90..8f25114 100644
--- a/tests/test_build.py
+++ b/tests/test_build.py
@@ -1,6 +1,9 @@
 import hashlib
+import contextlib
+import io
 import importlib.util
 import json
+import lzma
 import os
 from pathlib import Path
 import tempfile
@@ -109,6 +112,209 @@ class BuilderTests(unittest.TestCase):
         other = self.module.Builder(self.project, self.project / 'other-build', self.project / 'cache')
         self.assertNotEqual(self.builder().runroot, other.runroot)
 
+    def existing_builder(self, source=None, cache=None, layer=None, snapshot='old-snapshot'):
+        """A real executable models only the external Podman boundary."""
+        source = source or self.project
+        cache = cache or self.project / 'cache'
+        image = self.project / 'image'
+        image.mkdir(exist_ok=True)
+        inputs = {'stage3': {'url': 'https://example.test/stage.tar.xz',
+                            'sha256': hashlib.sha256(lzma.compress(b'original layer')).hexdigest()},
+                  'portage': {'filename': 'repo.tar.gz', 'commit': 'old-snapshot'}}
+        (image / 'sources.json').write_text(json.dumps(inputs))
+        builder = self.builder()
+        (builder.work / 'podman').mkdir(parents=True)
+        fixture = self.project / 'podman-fixture.json'
+        fixture.write_text(json.dumps({'Image': 'original-image-id',
+            'Mounts': [{'Source': str(source), 'Destination': '/hydra-source', 'RW': False},
+                       {'Source': str(cache), 'Destination': '/hydra-cache', 'RW': False}],
+            'RootFS': {'Layers': [layer or 'sha256:' + hashlib.sha256(b'original layer').hexdigest()]},
+            'snapshot': snapshot}))
+        executable = self.project / 'fixture-podman'
+        executable.write_text('''#!/usr/bin/env python3
+import json, pathlib, sys
+root = pathlib.Path(__file__).parent
+data = json.loads((root / 'podman-fixture.json').read_text())
+args = sys.argv[7:]
+if 'exists' in args:
+    sys.exit(0)
+if 'inspect' in args:
+    print(json.dumps([data]))
+elif args[0] == 'top':
+    print('HPID ARGS')
+    if data.get('active_job'):
+        print('123 /usr/bin/python3 /usr/lib/python-exec/python3/emerge --update package')
+elif args[0] == 'cp':
+    snapshot = data.get('snapshot_pending') if args[1].endswith('.hydra-snapshot-input') else data['snapshot']
+    if snapshot is None:
+        sys.exit(1)
+    pathlib.Path(args[-1]).write_text(snapshot + '\\n')
+elif args[0] == 'exec' and data.get('stopped'):
+    sys.exit(125)
+elif args[0] == 'exec' and 'packages.lock' in args[-1]:
+    sys.exit(75 if data.get('job_lock') else 0)
+elif args[0] == 'exec' and args[-1].startswith('cat '):
+    snapshot = data['snapshot'] or (data.get('snapshot_pending') if '.hydra-snapshot-input' in args[-1] else None)
+    if snapshot is None:
+        sys.exit(1)
+    print(snapshot)
+else:
+    if args[0] == 'start':
+        data['stopped'] = False
+        (root / 'podman-fixture.json').write_text(json.dumps(data))
+    with (root / 'mutations.jsonl').open('a') as stream:
+        stream.write(json.dumps(args) + '\\n')
+''')
+        executable.chmod(0o755)
+        builder.podman = str(executable)
+        with builder.locked():
+            (builder.cache / 'stage.tar.xz').write_bytes(lzma.compress(b'original layer'))
+        return builder, inputs
+
+    def assert_prepare_refused(self, builder, inputs, reason):
+        with patch.object(builder, 'fetch', return_value=inputs):
+            with self.assertRaisesRegex(self.module.BuildError, reason):
+                builder.prepare()
+        self.assertFalse((self.project / 'mutations.jsonl').exists(), 'refusal must precede container/config mutations')
+        self.assertNotIn('prepare', builder.manifest(), 'refusal must not record false provenance')
+
+    def test_other_worktree_mount_is_refused_before_preparation(self):
+        builder, inputs = self.existing_builder(source=self.project / 'other-worktree')
+        self.assert_prepare_refused(builder, inputs, 'source')
+
+    def test_other_cache_mount_is_refused_before_preparation(self):
+        builder, inputs = self.existing_builder(cache=self.project / 'other-cache')
+        self.assert_prepare_refused(builder, inputs, 'cache')
+
+    def test_changed_stage3_pin_refuses_old_container(self):
+        builder, inputs = self.existing_builder()
+        archive = lzma.compress(b'new stage layer')
+        (builder.cache / 'stage.tar.xz').write_bytes(archive)
+        inputs['stage3']['sha256'] = hashlib.sha256(archive).hexdigest()
+        self.assert_prepare_refused(builder, inputs, 'stage3')
+
+    def test_changed_snapshot_pin_refuses_existing_tree(self):
+        builder, inputs = self.existing_builder()
+        inputs['portage']['commit'] = 'new-snapshot'
+        self.assert_prepare_refused(builder, inputs, 'snapshot')
+
+    def test_matching_existing_builder_retains_compatible_provenance(self):
+        builder, inputs = self.existing_builder()
+        with patch.object(builder, 'fetch', return_value=inputs):
+            builder.prepare()
+        self.assertEqual(builder.manifest()['prepare'].get('stage3_sha256'), inputs['stage3']['sha256'])
+        self.assertEqual(builder.manifest()['prepare']['snapshot'], 'old-snapshot')
+
+    def test_status_invalidates_missing_package_inventory(self):
+        builder = self.builder()
+        with builder.locked():
+            builder.record('packages')
+        output = io.StringIO()
+        with contextlib.redirect_stdout(output):
+            builder.status()
+        self.assertFalse(json.loads(output.getvalue())['packages']['complete'])
+
+    def test_status_cli_runs_during_active_build_without_mutations(self):
+        builder = self.builder()
+        self.module.subprocess.run(['git', 'init', '--quiet', str(self.project)], check=True)
+        output = io.StringIO()
+        with builder.locked():
+            builder.record('packages')
+            manifest_before = (builder.work / 'manifest.json').read_bytes()
+            with patch.object(self.module, '__file__', str(self.project / 'hydra/build.py')):
+                with contextlib.redirect_stdout(output):
+                    result = self.module.main(['status'])
+            self.assertEqual(result, 0, 'status must not acquire the exclusive build lock')
+            data = json.loads(output.getvalue())
+            self.assertTrue(data['running'])
+            self.assertFalse(data['packages']['complete'])
+            self.assertEqual((builder.work / 'manifest.json').read_bytes(), manifest_before)
+
+    def test_detached_emerge_refuses_preparation_before_fetch_or_writes(self):
+        builder, inputs = self.existing_builder()
+        fixture = self.project / 'podman-fixture.json'
+        data = json.loads(fixture.read_text())
+        data['active_job'] = True
+        fixture.write_text(json.dumps(data))
+        with patch.object(builder, 'fetch', return_value=inputs) as fetch:
+            with self.assertRaisesRegex(self.module.BuildError, 'package job'):
+                builder.prepare()
+            fetch.assert_not_called()
+        self.assertFalse((self.project / 'mutations.jsonl').exists())
+
+    def test_contained_job_lock_refuses_preparation_before_fetch(self):
+        builder, inputs = self.existing_builder()
+        fixture = self.project / 'podman-fixture.json'
+        data = json.loads(fixture.read_text())
+        data['job_lock'] = True
+        fixture.write_text(json.dumps(data))
+        with patch.object(builder, 'fetch', return_value=inputs) as fetch:
+            with self.assertRaisesRegex(self.module.BuildError, 'package job'):
+                builder.prepare()
+            fetch.assert_not_called()
+
+    def test_status_detects_detached_emerge_without_outer_lock(self):
+        builder, inputs = self.existing_builder()
+        fixture = self.project / 'podman-fixture.json'
+        data = json.loads(fixture.read_text())
+        data['active_job'] = True
+        fixture.write_text(json.dumps(data))
+        output = io.StringIO()
+        with contextlib.redirect_stdout(output):
+            builder.status()
+        status = json.loads(output.getvalue())
+        self.assertTrue(status['running'])
+        self.assertTrue(status['package_job_running'])
+        self.assertFalse(status['mutation_lock_held'])
+
+    def test_package_job_records_failure_and_log_without_host_collection(self):
+        self.assertTrue(hasattr(self.module, 'package_job_script'), 'durable job runner is required')
+        script = self.module.package_job_script(['/bin/sh', '-c', 'echo diagnostic; exit 7'],
+                                               self.project / 'job', self.project / 'pkgdb', 'inputs-one')
+        result = self.module.subprocess.run(['python3', '-c', script])
+        self.assertEqual(result.returncode, 7)
+        status = json.loads((self.project / 'job/packages.status.json').read_text())
+        self.assertFalse(status['running'])
+        self.assertEqual(status['exit_code'], 7)
+        self.assertIn('diagnostic', (self.project / 'job/packages.log').read_text())
+        self.assertFalse((self.project / 'job/packages.json').exists())
+
+    def test_package_job_persists_inventory_after_success_without_host_collection(self):
+        self.assertTrue(hasattr(self.module, 'package_job_script'), 'durable job runner is required')
+        package = self.project / 'pkgdb/category/example-1.0'
+        package.mkdir(parents=True)
+        (package / 'USE').write_text('wayland opengl')
+        (package / 'BUILD_TIME').write_text('123')
+        script = self.module.package_job_script(['/bin/sh', '-c', 'exit 0'], self.project / 'job',
+                                               self.project / 'pkgdb', 'inputs-one')
+        result = self.module.subprocess.run(['python3', '-c', script])
+        self.assertEqual(result.returncode, 0)
+        self.assertEqual(json.loads((self.project / 'job/packages.json').read_text()),
+                         [{'package': 'category/example-1.0', 'use': ['wayland', 'opengl'], 'build_time': '123'}])
+        self.assertEqual(json.loads((self.project / 'job/packages.status.json').read_text())['exit_code'], 0)
+
+    def test_matching_stopped_builder_checks_provenance_before_start(self):
+        builder, inputs = self.existing_builder()
+        fixture = self.project / 'podman-fixture.json'
+        data = json.loads(fixture.read_text())
+        data['stopped'] = True
+        fixture.write_text(json.dumps(data))
+        with patch.object(builder, 'fetch', return_value=inputs):
+            try:
+                builder.prepare()
+            except self.module.BuildError as error:
+                self.fail(f'Compatible stopped builder must resume: {error}')
+        self.assertTrue(builder.manifest()['prepare']['complete'])
+
+    def test_unfinished_snapshot_is_not_reported_as_prepared(self):
+        builder, inputs = self.existing_builder(snapshot=None)
+        fixture = self.project / 'podman-fixture.json'
+        data = json.loads(fixture.read_text())
+        data['snapshot_pending'] = 'old-snapshot'
+        fixture.write_text(json.dumps(data))
+        builder.record('prepare')
+        self.assertFalse(builder.complete('prepare'))
+
 
 if __name__ == '__main__':
     unittest.main()