feat(oci): support custom Compose workers and migration jobs

2026-11-02 10:52:22 +00:00 · 2026-10-28 08:41:41 +01:00
parent b383fb9311
commit 3a0a7eaf14
44 changed files with 3267 additions and 110 deletions
@@ -96,7 +96,7 @@ def report_error(text, tail=True):

 def persist_instances(deployment, services, primary_id=None):
    template = json.loads(Path(sys.argv[1]).read_text()) if primary_id is not None else {}
-    with oci_instances.locked(oci_instances.ROOT):
+    with oci_instances.locked(oci_instances.ROOT, timeout=60):
        if primary_id is not None:
            oci_instances.save_assembly(oci_instances.ROOT, primary_id, template, deployment, services)
            oci_instances.resume_assembly(oci_instances.ROOT, primary_id)
@@ -244,6 +244,31 @@ def relay_child_output(stream):
    return result


+def assign_health_address(service):
+    """Bind a generated dependency probe to its allocated private address."""
+    check = service['healthcheck']
+    if check['type'] == 'http':
+        endpoint = check['endpoint']
+        check['url'] = f"{endpoint['scheme']}://{service['ip']}:{endpoint['port']}{endpoint['path']}"
+    elif check['type'] == 'tcp':
+        check['host'] = service['ip']
+
+
+def bind_completed_jobs(services):
+    """All members and private aliases exist before a job can be authorized."""
+    from oci_compose_jobs import job_spec_from_record
+    for service in services:
+        check = service['healthcheck']
+        if check.get('type') != 'completed':
+            continue
+        vmid = int(service['vmid'])
+        run('pct', 'set', vmid, '--onboot', '0')
+        service['deployment']['onboot'] = False
+        record = oci_instances.read(oci_instances.ROOT, vmid)
+        check['job'] = job_spec_from_record(record, run('pct', 'config', vmid),
+                                            int(check.get('timeout_seconds', 300)))
+
+
 def lan_access_urls(services, subnet, strict=True):
    urls = []
    for service in services:
@@ -271,15 +296,43 @@ def lan_access_urls(services, subnet, strict=True):
    return urls


-def _last_int_env(service, names):
-    """Last matching PUID/PGID-style value from deployment.environment,
-    or 0 if absent/unparsable — mirrors install_oci.sh's
-    `[.environment[]? | select(.name == ...) | .value] | last // "0"`
-    plus its numeric-guard `[[ $X =~ ^[0-9]+$ ]] || X=0`."""
+def wait_main_web(service, subnet):
+    """Check the advertised frontend, preserving private IPs for backend aliases.
+
+    A service may legitimately refuse HTTP requests whose Host is its private
+    stack address. The endpoint shown to the user is the readiness contract.
+    """
+    deadline = time.monotonic() + min(120, int(service['healthcheck']['timeout_seconds']))
+    url = None
+    while time.monotonic() < deadline:
+        try:
+            found = lan_access_urls([service], subnet)
+            if found:
+                url = found[0]['url']
+                break
+        except (RuntimeError, OSError, subprocess.SubprocessError):
+            pass
+        time.sleep(1)
+    if url is None:
+        raise RuntimeError(translate('No LAN address was obtained'))
+    service['healthcheck']['url'] = url
+    wait_web(service, url)
+    return url
+
+
+def _last_int_env(service, names, default=0):
+    """Use the last explicit environment override, falling back only if absent.
+
+    Invalid explicit values retain the installer's existing zero fallback.
+    """
    value = None
+    found = False
    for entry in service['deployment'].get('environment', []):
        if entry.get('name') in names:
+            found = True
            value = entry.get('value')
+    if not found:
+        return default
    try:
        parsed = int(value)
    except (TypeError, ValueError):
@@ -287,28 +340,78 @@ def _last_int_env(service, names):
    return parsed if parsed >= 0 else 0


-def resolve_volume_owner(service, owner_strategy):
-    """(uid, gid) for a given volume_preparations owner_strategy,
-    mirroring install_oci.sh's HOST_ROOT_UID/HOST_ROOT_GID (mapped-root)
-    and HOST_BIND_UID/HOST_BIND_GID (mapped-application-user) — these
-    are deliberately DIFFERENT values, not the same computation:
+def runtime_volume_owner(service, root=None):
+    """Container IDs from the OCI import, with image USER as a fallback.

-    - mapped-root: 100000:100000 for an unprivileged container, 0:0
-      otherwise. Never depends on PUID/PGID or volume_owner (matches
-      install_oci.sh's HOST_ROOT_UID/HOST_ROOT_GID, which are pinned
-      constants that do not read CONTAINER_PUID/CONTAINER_PGID at all).
-    - mapped-application-user: the container's own PUID/PGID (last of
-      PUID/USER_ID/UID and PGID/GROUP_ID/GID in deployment.environment,
-      overridden by the template's volume_owner.uid/gid when present —
-      matches install_oci.sh's IMAGE_VOLUME_UID/GID override of
-      CONTAINER_PUID/CONTAINER_PGID before the offset is applied),
-      offset by 100000 for an unprivileged container (matches
-      HOST_BIND_UID/HOST_BIND_GID). volume_owner only ever overrides
-      this application-user value, never mapped-root's.
+    Proxmox has already resolved a named USER and Compose user override into
+    lxc.init.uid/gid. No ownership is changed here. The mounted image account
+    files are only needed on versions that omit one of those numeric fields.
+    """
+    vmid = service.get('vmid')
+    config = {}
+    if type(vmid) is int and vmid >= 100:
+        from oci_installation_state import parse_config
+        config = parse_config(run('pct', 'config', vmid).encode())
+    resolved = {}
+    for name, key in (('uid', 'lxc.init.uid'), ('gid', 'lxc.init.gid')):
+        value = config.get(key)
+        if value is not None:
+            if not re.fullmatch(r'[0-9]+', value):
+                raise RuntimeError(translate('The imported OCI user or group is not numeric'))
+            resolved[name] = int(value)
+    if len(resolved) == 2:
+        return resolved['uid'], resolved['gid']
+    image = (service.get('observed') or {}).get('image') or {}
+    if not image and type(vmid) is int and vmid >= 100:
+        try:
+            record = oci_instances.read(oci_instances.ROOT, vmid)
+            image = (record.get('observed') or {}).get('image') or {}
+        except (OSError, ValueError, KeyError):
+            pass
+    declared = str((image.get('defaults') or {}).get('User') or '')
+    user, separator, group = declared.partition(':')
+    accounts, groups = {}, {}
+    if root is not None:
+        root = Path(root)
+        for filename, target in (('passwd', accounts), ('group', groups)):
+            path = root / 'etc' / filename
+            if path.resolve().is_relative_to(root.resolve()) and path.is_file() and not path.is_symlink():
+                for line in path.read_text(errors='replace').splitlines():
+                    fields = line.split(':')
+                    if len(fields) >= (4 if filename == 'passwd' else 3) and fields[2].isdigit():
+                        if filename == 'passwd' and fields[3].isdigit():
+                            target[fields[0]] = (int(fields[2]), int(fields[3]))
+                        elif filename == 'group':
+                            target[fields[0]] = int(fields[2])
+    if not user:
+        uid, gid = 0, 0
+    elif user.isdigit():
+        uid = int(user)
+        gid = next((value[1] for value in accounts.values() if value[0] == uid), 0)
+    elif user in accounts:
+        uid, gid = accounts[user]
+    elif 'uid' in resolved:
+        uid = resolved['uid']
+        gid = next((value[1] for value in accounts.values() if value[0] == uid), 0)
+    else:
+        raise RuntimeError(translate('The imported OCI user or group is not numeric'))
+    if separator and group:
+        if group.isdigit():
+            gid = int(group)
+        elif group in groups:
+            gid = groups[group]
+        elif 'gid' not in resolved:
+            raise RuntimeError(translate('The imported OCI user or group is not numeric'))
+    return resolved.get('uid', uid), resolved.get('gid', gid)

-    Any value that cannot be parsed as a non-negative integer is
-    treated as 0, matching install_oci.sh's `[[ ... =~ ^[0-9]+$ ]] ||
-    ...=0` guard.
+
+def resolve_volume_owner(service, owner_strategy, imported_owner=None):
+    """Resolve fresh volume ownership without changing existing host directories.
+
+    Mapped root remains fixed. Application ownership starts with the actual
+    OCI import user, then honors explicit PUID/PGID aliases and the recipe's
+    declared volume_owner override. All values are container IDs until the
+    standard unprivileged offset is applied exactly once.
    """
    unprivileged = bool(service['deployment']['security']['unprivileged'])
    if owner_strategy == 'mapped-root':
@@ -316,8 +419,9 @@ def resolve_volume_owner(service, owner_strategy):
    if owner_strategy != 'mapped-application-user':
        raise RuntimeError(f"{translate('Unsupported owner strategy:')} {owner_strategy}")

-    puid = _last_int_env(service, ('PUID', 'USER_ID', 'UID'))
-    pgid = _last_int_env(service, ('PGID', 'GROUP_ID', 'GID'))
+    imported_uid, imported_gid = (runtime_volume_owner(service) if imported_owner is None else imported_owner)
+    puid = _last_int_env(service, ('PUID', 'USER_ID', 'UID'), imported_uid)
+    pgid = _last_int_env(service, ('PGID', 'GROUP_ID', 'GID'), imported_gid)

    volume_owner = service.get('template', {}).get('proxmox', {}) \
        .get('installer_profile', {}).get('volume_owner') or {}
@@ -338,7 +442,7 @@ def resolve_volume_owner(service, owner_strategy):
    return (puid + offset, pgid + offset)


-def apply_volume_preparation(service, mount, target):
+def apply_volume_preparation(service, mount, target, imported_owner=None):
    """Declarative volume_preparations for one already-attached (but
    still empty/freshly-mounted) mount, mirroring install_oci.sh's
    apply_installer_profile() semantics (only_when_mount_type /
@@ -381,7 +485,7 @@ def apply_volume_preparation(service, mount, target):
        if lost.is_dir() and not lost.is_symlink():
            lost.rmdir()
    owner_strategy = match.get('owner_strategy')
-    owner = resolve_volume_owner(service, owner_strategy)
+    owner = resolve_volume_owner(service, owner_strategy, imported_owner)
    log(LOG, f"Volume prepared before the first start: {mount['container_path']}")
    return True, owner

@@ -390,6 +494,7 @@ def attach_mounts(service, temporary):
    """Populate new managed volumes from the image, preserving its ownership."""
    vmid = service['vmid']
    root = Path(f'/var/lib/lxc/{vmid}/rootfs')
+    imported_owner = None
    for index, mount in enumerate(service['deployment']['mounts']):
        target = root / mount['container_path'].lstrip('/')
        seed = temporary / f'seed-{vmid}-{index}'
@@ -398,6 +503,8 @@ def attach_mounts(service, temporary):
        try:
            if not target.resolve().is_relative_to(root.resolve()) or target.is_symlink():
                raise RuntimeError(translate('Unsafe volume path'))
+            if imported_owner is None:
+                imported_owner = runtime_volume_owner(service, root)
            if target.is_dir():
                stat = target.stat()
                owner = (stat.st_uid,stat.st_gid,stat.st_mode & 0o777)
@@ -437,7 +544,7 @@ def attach_mounts(service, temporary):
        if mount['type']=='managed-volume':
            run('pct','mount',vmid)
            try:
-                applied, declared_owner = apply_volume_preparation(service, mount, target)
+                applied, declared_owner = apply_volume_preparation(service, mount, target, imported_owner)
                if not applied:
                    lost=target/'lost+found'
                    if lost.is_dir() and not lost.is_symlink():
@@ -472,6 +579,13 @@ def wait_web(primary, url):
        try:
            with opener.open(url,timeout=4) as response:
                if response.status<400:
+                    check = primary['healthcheck']
+                    if check.get('type') == 'exec':
+                        remaining = max(1, deadline-time.monotonic())
+                        process = subprocess.run(['pct', 'exec', str(primary['vmid']), '--', *check['argv']],
+                                                 capture_output=True, timeout=min(30, remaining), check=False)
+                        if process.returncode != 0:
+                            raise RuntimeError('Health check has not succeeded yet')
                    return
        except Exception:
            pass
@@ -744,6 +858,12 @@ def finish_independent_suite(services, subnet):
    return result


+def check_selinux_labels(template, enforcing_path=Path('/sys/fs/selinux/enforce')):
+    if (template.get('proxmox', {}).get('custom_compose_selinux_labels')
+            and enforcing_path.is_file() and enforcing_path.read_text().strip() == '1'):
+        raise RuntimeError(translate('SELinux volume labels cannot be ignored while SELinux is enforcing.'))
+
+
 def main():
    deployment = json.loads(Path(sys.argv[2]).read_text())
    if deployment.get('deployment_kind') != 'generic-multi-lxc-stack':
@@ -756,6 +876,7 @@ def main():
        return
    if os.geteuid()!=0:
        raise RuntimeError(translate('The installer must run as root on Proxmox VE'))
+    check_selinux_labels(json.loads(Path(sys.argv[1]).read_text()))
    init_log(deployment.get('stack_name') or 'stack')
    services = copy.deepcopy(deployment['services'])
    independent = bool(deployment.get('suite_arr'))
@@ -785,7 +906,7 @@ def main():
            s['vmid'] = base+s['offset']
            s['ip'] = str(subnet.network_address+30+s['offset'])
            for alias in s['aliases']:
-                if not re.fullmatch(r'[A-Za-z0-9][A-Za-z0-9_.-]*',alias):
+                if not re.fullmatch(r'[A-Za-z0-9][A-Za-z0-9_.-]{0,252}',alias):
                    raise RuntimeError(translate('Invalid service alias'))
                aliases.append({'hostname':alias,'address':s['ip']})
        if not independent:
@@ -837,17 +958,18 @@ def main():
                if s['main'] or s.get('frontend'):
                    address = access_address(s.get('frontend_ipv4') or 'dhcp', deployment['network'].get('frontend_gateway'))
                    run('pct','set',s['vmid'],'--net1',f"name=eth1,bridge={deployment['network']['frontend_bridge']},{address},host-managed=1,firewall=1,type=veth")
-                if s['healthcheck']['type']=='http':
-                    endpoint = s['healthcheck']['endpoint']
-                    s['healthcheck']['url'] = f"{endpoint['scheme']}://{s['ip']}:{endpoint['port']}{endpoint['path']}"
+                assign_health_address(s)
+            # Existing CT/network configurations now reserve every allocated ID and subnet.
+            # Never retain the allocation lock while startup/hooks wait for another registry operation.
+            fcntl.flock(lock,fcntl.LOCK_UN)
            if independent:
-                fcntl.flock(lock,fcntl.LOCK_UN)
                result = finish_independent_suite(services,subnet)
                persist_instances(deployment, services)
                refresh_notes(services)
                result.update(completion_notes=list(deployment.get('completion_notes',[])), log=LOG)
                print('PROXMENUX_RESULT='+base64.b64encode(json.dumps(result).encode()).decode(),flush=True)
                return
+            bind_completed_jobs(services)
            hook_spec = Path(tmp)/'lifecycle.json'
            write_json(hook_spec,{'schema':1,'stack':deployment['stack_name'],'dependencies':[
                {'vmid':s['vmid'],'label':s['name'],'healthcheck':s['healthcheck']} for s in services if not s['main'] and not s.get('deferred_setup')]})
@@ -855,20 +977,15 @@ def main():
                msg_info(translate('Installing the stack startup hook...'))
                run('bash',HERE/'stack_dependency_hook.sh','--install',primary_id,hook_spec,capture=False)
                msg_ok(translate('Stack startup hook installed'))
-            fcntl.flock(lock,fcntl.LOCK_UN)
            msg_info(translate('Starting the main container and its dependencies...') if len(services) > 1
                     else translate('Starting the container...'))
            run('pct','start',primary_id,capture=False,timeout=600)
            msg_ok(f"{translate('Container started')}: CT {primary_id} ({primary['name']})")
            endpoint = primary['healthcheck']['endpoint']
-            url = f"{endpoint['scheme']}://{primary['ip']}:{endpoint['port']}{endpoint['path']}"
            msg_info(translate('Waiting for the application to respond...'))
-            wait_web(primary,url)
-            addresses=run('lxc-info','-n',primary_id,'-iH').splitlines()
-            lan = next((a for a in addresses if re.fullmatch(r'\d+\.\d+\.\d+\.\d+',a) and ipaddress.ip_address(a) not in subnet),None)
-            if not lan:
-                raise RuntimeError(translate('No LAN address was obtained'))
-            public_url = f"{endpoint['scheme']}://{lan}:{endpoint['port']}{endpoint['path']}"
+            public_url = wait_main_web(primary, subnet)
+            url = public_url
+            lan = urllib.parse.urlsplit(public_url).hostname
            msg_ok(f"{translate('Application responding:')} {public_url}")
            if primary['public_environment']:
                msg_info(translate('Applying the LAN address to the application URLs...'))
@@ -994,6 +994,11 @@ prepare_nvidia_driver_links() {

 source "$SCRIPT_DIR/oci_nvidia_setup.sh"

+oci_valid_extra_hostname() {
+  # Compose network aliases may contain underscores; each entry stays one hosts-file token.
+  [[ ${#1} -le 253 && $1 =~ ^[A-Za-z0-9][A-Za-z0-9_.-]*$ ]]
+}
+
 apply_extra_hosts() {
  local count rootfs hosts_file temporary address hostname bridge_address failed=0
  count=$(jq '.extra_hosts? // [] | length' "$DEPLOYMENT_FILE")
@@ -1021,7 +1026,7 @@ apply_extra_hosts() {
          || { oci_log "Could not resolve host-gateway on $BRIDGE"; failed=1; break; }
        address=$bridge_address
      fi
-      [[ $hostname =~ ^[A-Za-z0-9][A-Za-z0-9.-]*$ ]] \
+      oci_valid_extra_hostname "$hostname" \
        || { oci_log "Invalid extra hostname: $hostname"; failed=1; break; }
      [[ $address =~ ^[0-9A-Fa-f:.]+$ ]] \
        || { oci_log "Invalid extra host address: $address"; failed=1; break; }
@@ -0,0 +1,479 @@
+#!/usr/bin/env python3
+"""Verify native OCI one-shot completion using authenticated raw LXC events.
+
+No image wrapper, shell injection or STOPPED-state inference is used. The
+supported monitor ABI was verified against upstream LXC v7.0.0 monitor.h,
+monitor.c and tools/lxc_monitor.c. The textual CLI loses signal information.
+This standalone helper is copied beside the persistent dependency hook.
+"""
+from __future__ import annotations
+
+import argparse
+import contextlib
+import datetime
+import fcntl
+import hashlib
+import json
+import os
+from pathlib import Path
+import re
+import select
+import socket
+import stat
+import struct
+import subprocess
+import tempfile
+import time
+import urllib.parse
+import uuid
+
+STATE_ROOT = Path('/var/lib/proxmenux/oci/jobs')
+INSTANCE_ROOT = Path('/usr/local/share/proxmenux/oci/instances')
+PCT = '/usr/sbin/pct'
+MONITOR = '/usr/bin/lxc-monitor'
+LXC_PATH = '/var/lib/lxc'
+FRAME = struct.Struct('@i256si')
+SUPPORTED_VERSIONS = frozenset({'7.0.0'})
+DIGEST = re.compile(r'sha256:[0-9a-f]{64}\Z')
+
+
+class JobError(RuntimeError):
+    pass
+
+
+def _vmid(value):
+    if type(value) is not int or not 100 <= value <= 999999999:
+        raise JobError('Invalid OCI job VMID')
+    return value
+
+
+def _root():
+    if os.geteuid() != 0:
+        raise JobError('OCI job execution requires root')
+
+
+def _trusted_read(path, *, owner=0):
+    """Open a root-owned regular file without following its final symlink."""
+    fd = os.open(path, os.O_RDONLY | os.O_CLOEXEC | os.O_NOFOLLOW)
+    try:
+        info = os.fstat(fd)
+        if not stat.S_ISREG(info.st_mode) or info.st_uid != owner or info.st_mode & 0o022:
+            raise JobError('Untrusted OCI job metadata file')
+        if info.st_size > 8 * 1024 * 1024:
+            raise JobError('OCI job metadata file is too large')
+        with os.fdopen(fd, 'r', encoding='utf-8') as source:
+            fd = -1
+            return source.read()
+    finally:
+        if fd >= 0:
+            os.close(fd)
+
+
+def _private_directory(path):
+    path.mkdir(parents=True, mode=0o700, exist_ok=True)
+    info = path.lstat()
+    if (not stat.S_ISDIR(info.st_mode) or info.st_uid != 0 or info.st_mode & 0o077):
+        raise JobError('OCI job state directory is not private')
+
+
+def config_fingerprint(text):
+    """Presentation and resource changes do not re-run a completed migration."""
+    selected = []
+    for line in text.split('\n'):
+        if line.startswith('['):
+            break  # Pending/snapshot sections are not the active definition.
+        key, separator, value = line.partition(': ')
+        if not separator:
+            continue
+        if (key in {'entrypoint', 'env', 'rootfs', 'unprivileged', 'features', 'ostype'}
+                or re.fullmatch(r'(?:mp|dev)[0-9]+', key)
+                or key.startswith(('lxc.init.', 'lxc.environment.', 'lxc.mount.',
+                                   'lxc.cgroup', 'lxc.apparmor', 'lxc.seccomp'))):
+            if key == 'rootfs' or re.fullmatch(r'mp[0-9]+', key):
+                value = ','.join(part for part in value.split(',') if not part.startswith('size='))
+            selected.append((key, value))
+    if not any(key == 'rootfs' for key, _ in selected):
+        raise JobError('OCI job has no root filesystem configuration')
+    selected.sort(key=lambda item: item[0])
+    return hashlib.sha256(json.dumps(selected, separators=(',', ':'), ensure_ascii=True).encode()).hexdigest()
+
+
+def _config_text(vmid):
+    # Validate the actual PVE file's ownership; use the same API rendering
+    # whose exact hash the coordinated transaction authorizes.
+    _trusted_read(Path('/etc/pve/lxc') / f'{vmid}.conf')
+    result = subprocess.run([PCT, 'config', str(vmid)], capture_output=True, check=True, timeout=15)
+    return result.stdout.decode('utf-8')
+
+
+def _record_digest(record, config_text, *, executing=False):
+    pending = record.get('pending_transaction')
+    if not pending:
+        return record.get('observed', {}).get('resolved_registry_digest')
+    vmid = _vmid(record.get('vmid'))
+    if record.get('status') != 'updating' or not record.get('pending_stack_transaction'):
+        raise JobError('OCI job has no authorized coordinated update context')
+    journal = Path(pending)
+    try:
+        relative = journal.relative_to(INSTANCE_ROOT / str(vmid) / 'transactions')
+    except ValueError as error:
+        raise JobError('OCI job transaction is outside its private instance state') from error
+    if (len(relative.parts) != 2 or relative.parts[1] != 'transaction.json'
+            or not re.fullmatch(r'[0-9a-f]{32}', relative.parts[0])):
+        raise JobError('Invalid OCI job transaction path')
+    child = json.loads(_trusted_read(journal))
+    coordinated = child.get('coordinated') or {}
+    if (child.get('schema_version') != 1 or child.get('id') != record.get('transaction_id')
+            or child.get('id') != relative.parts[0] or child.get('vmid') != vmid
+            or child.get('phase') not in {'candidate-installed', 'health-passed', 'committing'}
+            or child.get('record', {}).get('installation_id') != record.get('installation_id')
+            or coordinated.get('journal') != record.get('pending_stack_transaction')
+            or child.get('staged_config_sha256') != hashlib.sha256(config_text.encode()).hexdigest()):
+        raise JobError('OCI job staged image is not bound to this actual container configuration')
+    stack_journal = Path(record['pending_stack_transaction'])
+    try:
+        stack_path = stack_journal.relative_to(INSTANCE_ROOT)
+    except ValueError as error:
+        raise JobError('OCI job stack transaction is outside private instance state') from error
+    if (len(stack_path.parts) != 4 or not stack_path.parts[0].isdigit()
+            or stack_path.parts[1] != 'stack-transactions'
+            or not re.fullmatch(r'[0-9a-f]{32}', stack_path.parts[2])
+            or stack_path.parts[3] != 'transaction.json'):
+        raise JobError('Invalid OCI job stack transaction path')
+    stack = json.loads(_trusted_read(stack_journal))
+    phases = {'starting', 'checking', 'publishing', 'recovering'}
+    if not executing:
+        phases.add('replacing')
+    members = stack.get('plan', {}).get('members') or []
+    if (stack.get('schema_version') != 1 or stack.get('id') != coordinated.get('id')
+            or stack.get('phase') not in phases
+            or not any(member.get('vmid') == vmid
+                       and member.get('installation_id') == record.get('installation_id')
+                       for member in members)):
+        raise JobError('OCI job is not authorized by the active stack transaction')
+    return child.get('registry_digest')
+
+
+def job_spec_from_record(record, config_text, timeout_seconds=300):
+    """Bind a first installation, staged update or restored member for its hook."""
+    if isinstance(config_text, bytes):
+        config_text = config_text.decode('utf-8')
+    if not isinstance(config_text, str) or not isinstance(record, dict):
+        raise JobError('Invalid OCI job binding input')
+    if type(timeout_seconds) is not int or not 30 <= timeout_seconds <= 3600:
+        raise JobError('OCI job timeout must be between 30 and 3600 seconds')
+    _vmid(record.get('vmid'))
+    try:
+        installation_id = str(uuid.UUID(record['installation_id']))
+    except (KeyError, ValueError, TypeError, AttributeError) as error:
+        raise JobError('OCI job requires an installation identity') from error
+    digest = _record_digest(record, config_text)
+    if not isinstance(digest, str) or not DIGEST.fullmatch(digest):
+        raise JobError('OCI job requires a verified image digest')
+    return {'installation_id': installation_id, 'image_digest': digest,
+            'expected_config_fingerprint': config_fingerprint(config_text),
+            'timeout_seconds': timeout_seconds}
+
+
+def current_identity(vmid, spec):
+    """Bind the request to both a trusted instance record and the actual CT."""
+    _vmid(vmid)
+    if not isinstance(spec, dict):
+        raise JobError('Invalid OCI job specification')
+    try:
+        installation_id = str(uuid.UUID(spec['installation_id']))
+    except (KeyError, ValueError, TypeError, AttributeError) as error:
+        raise JobError('OCI job requires an installation identity') from error
+    digest = spec.get('image_digest')
+    if not isinstance(digest, str) or not DIGEST.fullmatch(digest):
+        raise JobError('OCI job requires a verified image digest')
+    record = json.loads(_trusted_read(INSTANCE_ROOT / str(vmid) / 'oci-compose.json'))
+    if record.get('vmid') != vmid or record.get('installation_id') != installation_id:
+        raise JobError('OCI job installation identity changed')
+    config = _config_text(vmid)
+    if _record_digest(record, config, executing=True) != digest:
+        raise JobError('OCI job image digest changed')
+    description = next((line.partition(': ')[2] for line in config.split('\n')
+                        if line.startswith('description: ')), '')
+    markers = re.findall(r'proxmenux-instance=([0-9a-fA-F-]{36})', urllib.parse.unquote(description))
+    if installation_id not in markers:
+        raise JobError('The CT does not carry the OCI job installation identity')
+    fingerprint = config_fingerprint(config)
+    if spec.get('expected_config_fingerprint', fingerprint) != fingerprint:
+        raise JobError('OCI job runtime configuration changed')
+    recovery = record.get('recovered_at', '')
+    if not isinstance(recovery, str):
+        raise JobError('Invalid OCI job recovery generation')
+    return {'vmid': vmid, 'installation_id': installation_id, 'image_digest': digest,
+            'config_fingerprint': fingerprint, 'recovery_generation': recovery}
+
+
+def _status(vmid):
+    result = subprocess.run([PCT, 'status', str(vmid)], capture_output=True, text=True, timeout=15)
+    value = result.stdout.strip()
+    if result.returncode or value not in {'status: stopped', 'status: running'}:
+        raise JobError('Cannot establish the OCI job container state')
+    return value.removeprefix('status: ')
+
+
+def _socket_name(lxcpath=LXC_PATH):
+    value = 14695981039346656037
+    for byte in ('lxc/'+lxcpath+'/monitor-sock').encode():
+        value = ((value ^ byte) * 1099511628211) & ((1 << 64)-1)
+    return f'lxc/{value:016x}/{lxcpath}'
+
+
+@contextlib.contextmanager
+def _subscription():
+    version = subprocess.run([MONITOR, '--version'], capture_output=True, text=True,
+                             check=True, timeout=10).stdout.strip()
+    if version not in SUPPORTED_VERSIONS or FRAME.size != 264:
+        raise JobError('Unsupported native LXC monitor ABI; refusing to guess job completion')
+    # This CLI only bootstraps monitord when no other container has started it.
+    # Its lossy textual exit status is NEVER consumed.
+    primer = subprocess.Popen([MONITOR, '-n', '^$'], stdin=subprocess.DEVNULL,
+                              stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
+                              close_fds=True)
+    receiver = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
+    try:
+        receiver.settimeout(1)
+        deadline = time.monotonic()+5
+        name = _socket_name()
+        while True:
+            try:
+                receiver.connect(b'\0'+name.encode())
+                break
+            except (ConnectionRefusedError, FileNotFoundError):
+                if time.monotonic() >= deadline:
+                    raise JobError('Native LXC monitor subscription could not be established')
+                time.sleep(.05)
+        peer = struct.unpack('3i', receiver.getsockopt(socket.SOL_SOCKET, socket.SO_PEERCRED,
+                                                      struct.calcsize('3i')))
+        pid, uid, gid = peer
+        executable = str(Path(f'/proc/{pid}/exe').resolve())
+        if uid != 0 or gid != 0 or executable not in {
+                '/usr/libexec/lxc/lxc-monitord', '/usr/lib/lxc/lxc-monitord'}:
+            raise JobError('Native LXC monitor peer is not the trusted root daemon')
+        receiver.setblocking(False)
+        yield receiver, {'version': version, 'peer_uid': uid, 'peer_gid': gid,
+                         'peer_executable': executable, 'protocol': 'lxc-msg-native264'}
+    finally:
+        receiver.close()
+        primer.terminate()
+        try:
+            primer.wait(timeout=3)
+        except subprocess.TimeoutExpired:
+            primer.kill()
+            primer.wait(timeout=3)
+
+
+def _frames(buffer):
+    messages = []
+    while len(buffer) >= FRAME.size:
+        chunk, buffer = buffer[:FRAME.size], buffer[FRAME.size:]
+        kind, name, value = FRAME.unpack(chunk)
+        if kind not in {0, 1, 2}:
+            raise JobError('Unknown native LXC monitor event')
+        try:
+            name = name.split(b'\0', 1)[0].decode('ascii')
+        except UnicodeDecodeError as error:
+            raise JobError('Malformed native LXC monitor event') from error
+        messages.append((kind, name, value))
+    return messages, buffer
+
+
+def _run_attempt(vmid, deadline):
+    with _subscription() as (receiver, proof):
+        # Drain traffic before our start; events from unrelated guests do not
+        # become evidence for this job. There is no historical event replay.
+        pending = b''
+        while select.select([receiver], [], [], 0)[0]:
+            data = receiver.recv(8192)
+            if not data:
+                raise JobError('Native LXC monitor closed before job start')
+            messages, pending = _frames(pending+data)
+            if any(name == str(vmid) for _, name, _ in messages):
+                raise JobError('OCI job changed state before its start attempt')
+        started = False
+        # Fast successful PID1 jobs can make pct start return 255 because
+        # Proxmox can no longer query the PID. Only raw exit evidence decides.
+        process = subprocess.Popen([PCT, 'start', str(vmid)], stdin=subprocess.DEVNULL,
+                                   stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
+                                   close_fds=True)
+        try:
+            while time.monotonic() < deadline:
+                if not select.select([receiver], [], [], min(1, max(.01, deadline-time.monotonic())))[0]:
+                    continue
+                data = receiver.recv(8192)
+                if not data:
+                    raise JobError('Native LXC monitor closed without a job exit event')
+                messages, pending = _frames(pending+data)
+                for kind, name, value in messages:
+                    if name != str(vmid):
+                        continue
+                    if kind == 0 and value in {1, 2}:
+                        started = True
+                    if kind != 2:
+                        continue
+                    if not started or not 0 <= value <= 65535:
+                        raise JobError('No fresh, valid native OCI job exit proof')
+                    if not os.WIFEXITED(value):
+                        reason = f'signal {os.WTERMSIG(value)}' if os.WIFSIGNALED(value) else 'unknown wait status'
+                        raise JobError(f'OCI job CT {vmid} did not complete normally ({reason})')
+                    exitcode = os.WEXITSTATUS(value)
+                    if exitcode != 0:
+                        raise JobError(f'OCI job CT {vmid} exited with code {exitcode}')
+                    while _status(vmid) != 'stopped':
+                        if time.monotonic() >= deadline:
+                            raise JobError('OCI job reported exit zero but did not stop')
+                        time.sleep(.05)
+                    process.wait(timeout=max(.1, min(5, deadline-time.monotonic())))
+                    return {**proof, 'raw_waitstatus': value, 'exit_code': exitcode,
+                            'fresh_start_observed': True, 'pct_start_returncode': process.returncode}
+            raise JobError('OCI job timed out without verified native exit zero')
+        finally:
+            if process.poll() is None:
+                process.terminate()
+                try:
+                    process.wait(timeout=3)
+                except subprocess.TimeoutExpired:
+                    process.kill()
+                    process.wait(timeout=3)
+
+
+def _cache_matches(cache, identity):
+    return (isinstance(cache, dict) and cache.get('schema') == 1
+            and cache.get('status') == 'completed' and cache.get('identity') == identity
+            and isinstance(cache.get('proof'), dict)
+            and type(cache['proof'].get('raw_waitstatus')) is int
+            and cache['proof']['raw_waitstatus'] == 0
+            and cache['proof'].get('fresh_start_observed') is True)
+
+
+def _read_cache(path):
+    try:
+        return json.loads(_trusted_read(path))
+    except FileNotFoundError:
+        return None
+
+
+def _sync_directory(path):
+    descriptor = os.open(path, os.O_RDONLY | os.O_DIRECTORY | os.O_CLOEXEC)
+    try:
+        os.fsync(descriptor)
+    finally:
+        os.close(descriptor)
+
+
+def _write_cache(path, value):
+    descriptor, name = tempfile.mkstemp(prefix='.'+path.name+'.', dir=path.parent)
+    try:
+        with os.fdopen(descriptor, 'w', encoding='utf-8') as output:
+            json.dump(value, output, sort_keys=True)
+            output.write('\n')
+            output.flush()
+            os.fsync(output.fileno())
+        os.replace(name, path)
+        _sync_directory(path.parent)
+    finally:
+        with contextlib.suppress(FileNotFoundError):
+            os.unlink(name)
+
+
+@contextlib.contextmanager
+def _job_lock(vmid, deadline):
+    _private_directory(STATE_ROOT)
+    fd = os.open(STATE_ROOT/f'{vmid}.lock', os.O_CREAT|os.O_RDWR|os.O_NOFOLLOW|os.O_CLOEXEC, 0o600)
+    try:
+        info = os.fstat(fd)
+        if not stat.S_ISREG(info.st_mode) or info.st_uid != 0 or info.st_mode & 0o077:
+            raise JobError('Untrusted OCI job lock file')
+        while True:
+            try:
+                fcntl.flock(fd, fcntl.LOCK_EX|fcntl.LOCK_NB)
+                break
+            except BlockingIOError:
+                if time.monotonic() >= deadline:
+                    raise JobError('Timed out waiting for another OCI job attempt')
+                time.sleep(.05)
+        yield
+    finally:
+        os.close(fd)
+
+
+def ensure_completed(vmid, job_spec, force=False):
+    _root()
+    _vmid(vmid)
+    if type(force) is not bool:
+        raise JobError('Invalid OCI job force flag')
+    timeout = job_spec.get('timeout_seconds', 300) if isinstance(job_spec, dict) else None
+    if type(timeout) is not int or not 30 <= timeout <= 3600:
+        raise JobError('OCI job timeout must be between 30 and 3600 seconds')
+    deadline = time.monotonic()+timeout
+    with _job_lock(vmid, deadline):
+        identity = current_identity(vmid, job_spec)
+        if _status(vmid) != 'stopped':
+            raise JobError('OCI job is already running without this attempt\'s exit subscription')
+        path = STATE_ROOT/f'{vmid}.json'
+        cache = _read_cache(path)
+        if not force and _cache_matches(cache, identity):
+            return {'status': 'completed', 'cached': True, **identity, 'proof': cache['proof']}
+        # A forced failed retry must never leave an earlier success reusable.
+        if path.exists():
+            path.unlink()
+            _sync_directory(path.parent)
+        proof = _run_attempt(vmid, deadline)
+        if current_identity(vmid, job_spec) != identity:
+            raise JobError('OCI job identity or configuration changed during execution')
+        if _status(vmid) != 'stopped':
+            raise JobError('OCI job restarted before completion could be recorded')
+        value = {'schema': 1, 'status': 'completed', 'identity': identity, 'proof': proof,
+                 'attempt_id': str(uuid.uuid4()),
+                 'completed_at': datetime.datetime.now(datetime.timezone.utc).isoformat()}
+        _write_cache(path, value)
+        return {'status': 'completed', 'cached': False, **identity, 'proof': proof}
+
+
+def forget_completed(vmid, installation_id):
+    """Remove only this installation's completion, never a reused VMID's state."""
+    _vmid(vmid)
+    try:
+        installation_id = str(uuid.UUID(installation_id))
+    except (ValueError, TypeError, AttributeError):
+        return False  # Older non-job records may use a legacy identity format.
+    if not STATE_ROOT.exists():
+        return False
+    _root()
+    with _job_lock(vmid, time.monotonic()+30):
+        path = STATE_ROOT/f'{vmid}.json'
+        cache = _read_cache(path)
+        if (not isinstance(cache, dict) or not isinstance(cache.get('identity'), dict)
+                or cache['identity'].get('installation_id') != installation_id):
+            return False
+        path.unlink()
+        _sync_directory(path.parent)
+        # The stable lock inode stays: unlinking it could split concurrent
+        # waiters across two locks when this numeric VMID is reused.
+        return True
+
+
+def main():
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument('operation', choices=['ensure'])
+    parser.add_argument('vmid', type=int)
+    parser.add_argument('--spec-json', required=True)
+    parser.add_argument('--force', action='store_true')
+    args = parser.parse_args()
+    try:
+        if len(args.spec_json) > 65536:
+            raise JobError('OCI job specification is too large')
+        result = ensure_completed(args.vmid, json.loads(args.spec_json), force=args.force)
+    except (JobError, OSError, ValueError, KeyError, subprocess.SubprocessError) as error:
+        print(str(error), file=__import__('sys').stderr)
+        return 1
+    print(json.dumps(result, sort_keys=True))
+    return 0
+
+
+if __name__ == '__main__':
+    raise SystemExit(main())
@@ -15,6 +15,7 @@ import re
 import subprocess
 import sys
 import tempfile
+import time
 import uuid

 from oci_installation_state import command, image_from_archive, private_directory, sha
@@ -42,8 +43,22 @@ def location(root, vmid):
    return path / 'oci-compose.json'


+def _acquire_lock(fd, timeout):
+    """Wait only for acquisition; a failure inside the operation is never replayed."""
+    deadline = time.monotonic() + max(0, float(timeout))
+    while True:
+        try:
+            fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB)
+            return
+        except BlockingIOError:
+            remaining = deadline - time.monotonic()
+            if remaining <= 0:
+                raise
+            time.sleep(min(0.2, remaining))
+
+
@contextmanager
-def locked(root):
+def locked(root, timeout=0):
    private_directory(root)
    path = root / '.lock'
    if path.is_symlink():
@@ -53,12 +68,12 @@ def locked(root):
        fd = int(inherited)
        if os.fstat(fd).st_ino != path.stat().st_ino or os.fstat(fd).st_dev != path.stat().st_dev:
            raise ValueError(translate('Wrong inherited registry lock'))
-        fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB)
+        _acquire_lock(fd, timeout)
        yield
    else:
        with path.open('a') as lock:
            os.chmod(path, 0o600)
-            fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB)
+            _acquire_lock(lock, timeout)
            yield


@@ -19,6 +19,7 @@ import sys
 import oci_image_cache as image_cache
 import oci_work_backup as work_backup
 import oci_instances as instances
+import oci_compose_jobs as compose_jobs
 from oci_installation_state import parse_config
 import oci_console
 import oci_runtime_settings as runtime_settings
@@ -365,6 +366,8 @@ def remove(root, vmid):
    if lifecycle.exists() and not lifecycle.is_symlink():
        lifecycle.unlink()
    for member in members:
+        record = instances.read(root, member)
+        compose_jobs.forget_completed(member, record['installation_id'])
        directory = instances.location(root, member).parent
        work_backup.forget(root, member)
        if directory.is_dir() and not directory.is_symlink():
@@ -32,6 +32,8 @@ import tempfile
 import oci_carried_record as carried
 import oci_console
 import oci_instances as instances
+import oci_compose_jobs as compose_jobs
+import oci_stack_plan
 import oci_instance_transaction as transaction
 import oci_nested_mounts
 import oci_nvidia_refresh
@@ -660,6 +662,19 @@ def valid_contract(contract, dependents):
                    return False
            except ValueError:
                return False
+        elif check.get('type') == 'tcp':
+            try:
+                if (not isinstance(check.get('host'), str)
+                        or ipaddress.IPv4Address(check['host']) not in PRIVATE
+                        or type(check.get('port')) is not int or not 0 < check['port'] < 65536):
+                    return False
+            except ValueError:
+                return False
+        elif check.get('type') == 'completed':
+            try:
+                oci_stack_plan.validate_job_check(check)
+            except (ValueError, TypeError, KeyError):
+                return False
        elif check.get('type') != 'running':
            return False
    return True
@@ -903,6 +918,7 @@ def register(root, plan):
                # leaves it; this also proves the stack can still be updated.
                oci_stack_modify.register(root, vmid)
        refresh_service_plans(root, [vmid for vmid, stack in written if stack])
+        refresh_completed_jobs(root, plan)
    except BaseException:
        for vmid, _ in written:
            path = instances.location(root, vmid)
@@ -945,6 +961,65 @@ def refresh_service_plans(root, vmids):
            instances.write(instances.location(root, vmid), record)


+def refresh_completed_jobs(root, plan):
+    """Rebind restored job proofs before a start, including a deferred start."""
+    primaries = set()
+    for entry in plan['restored']:
+        record = instances.read(root, entry['vmid'])
+        primary = record.get('stack_member', {}).get('primary_vmid')
+        if record.get('stack'):
+            primary = record['vmid']
+        if primary is not None:
+            primaries.add(primary)
+    for primary_id in primaries:
+        primary = instances.read(root, primary_id)
+        services = primary.get('stack', {}).get('deployment', {}).get('services', [])
+        jobs = [service for service in services if service.get('healthcheck', {}).get('type') == 'completed']
+        if not jobs:
+            continue
+        for service in jobs:
+            vmid = service['vmid']
+            record = instances.read(root, vmid)
+            if (record['deployment'].get('onboot') or record['deployment'].get('watchdog')
+                    or record['deployment'].get('execution_mode') != 'one-shot'):
+                raise ValueError(translate('Invalid service check arguments'))
+            config = checked('pct', 'config', str(vmid))
+            # Any restored member may include database data from before the cached proof.
+            record['recovered_at'] = instances.now()
+            instances.write(instances.location(root, vmid), record)
+            timeout = service['healthcheck'].get('timeout_seconds', 300)
+            service['healthcheck'] = {'type': 'completed', 'timeout_seconds': timeout,
+                                     'job': compose_jobs.job_spec_from_record(record, config, timeout)}
+        dependencies = [{'vmid': service['vmid'], 'label': service['name'],
+                         'healthcheck': service['healthcheck']} for service in services
+                        if service['vmid'] != primary_id and not service.get('deferred_setup')]
+        contract = {'schema': 1, 'stack': primary['stack']['deployment'].get('stack_name', ''),
+                    'dependencies': dependencies}
+        with tempfile.NamedTemporaryFile('w', suffix='.json') as file:
+            json.dump(contract, file)
+            file.flush()
+            checked('bash', str(ENGINE / 'stack_dependency_hook.sh'), '--install', str(primary_id), file.name)
+        for entry in plan['restored']:
+            if entry['vmid'] == primary_id:
+                entry['contract'] = contract
+        primary['stack']['deployment']['services'] = services
+        members = primary['stack'].get('members', [])
+        for index, member in enumerate(members):
+            if any(service['vmid'] == member.get('vmid') for service in jobs):
+                current = copy.deepcopy(instances.read(root, member['vmid']))
+                current.pop('stack', None)
+                members[index] = current
+        instances.write(instances.location(root, primary_id), primary)
+
+
+def completed_job(root, vmid):
+    record = instances.read(root, vmid)
+    primary = record.get('stack_member', {}).get('primary_vmid', vmid)
+    main = record if primary == vmid else instances.read(root, primary)
+    return any(service.get('vmid') == vmid and service.get('healthcheck', {}).get('type') == 'completed'
+               for service in main.get('stack', {}).get('deployment', {}).get('services', []))
+
+
 def start_order(entry):
    order = re.search(r'^startup: .*?order=([0-9]+)', entry['config'], re.MULTILINE)
    return int(order[1]) if order else 9999
@@ -990,6 +1065,9 @@ def recover(root, plan, start=False, host_firewall=()):
            msg_warn(f"{translate('Could not add the confirmed host firewall rule')}: {text}")
    if start and not plan['hold']:
        for entry in sorted(plan['restored'], key=start_order):
+            if completed_job(root, entry['vmid']):
+                # The main hook executes jobs in graph order with authenticated exit proofs.
+                continue
            if 'running' in run('pct', 'status', str(entry['vmid'])).stdout:
                continue
            msg_info(f"{translate('Starting the container...')} CT {entry['vmid']}")
@@ -3,6 +3,7 @@
 import argparse
 import copy
 import json
+import ipaddress
 import os
 from pathlib import Path
 import socket
@@ -12,6 +13,7 @@ import time
 import uuid

 import oci_image_cache as image_cache
+import oci_compose_jobs as compose_jobs
 import oci_instances as instances
 import oci_instance_transaction as member_tx
 import oci_stack_plan
@@ -23,6 +25,17 @@ from oci_update_current import resolve_archive
 from oci_ui import translate, msg_info, msg_ok, msg_warn, msg_error


+def tcp_target(check):
+    """Saved dependency probes only connect to numeric addresses and real TCP ports."""
+    host, port = check.get('host'), check.get('port')
+    if not isinstance(host, str) or type(port) is not int or not 0 < port < 65536:
+        raise ValueError(translate('Invalid service check arguments'))
+    try:
+        return str(ipaddress.IPv4Address(host)), port
+    except ValueError as error:
+        raise ValueError(translate('Invalid service check arguments')) from error
+
+
 def validate_database_transition(previous, candidate):
    def major(image):
        values = image.get('defaults', {}).get('Env') or []
@@ -206,8 +219,29 @@ class NativeAdapter:
            if pending and pending != str(self.journal):
                raise ValueError(translate('Another stack operation is pending'))

+    def _migration_backups(self):
+        if not any(self.is_job(vmid) for vmid in self.services):
+            return
+        database_kinds = {'postgres', 'mariadb', 'linuxserver/mariadb', 'mongo',
+                          'getmeili/meilisearch', 'redis', 'valkey/valkey'}
+        markers = ('PG_MAJOR=', 'MYSQL_MAJOR=', 'MYSQL_VERSION=', 'MARIADB_MAJOR=',
+                   'MARIADB_VERSION=', 'MONGO_VERSION=', 'MONGODB_VERSION=')
+        for vmid, record in self.records.items():
+            service = self.services[vmid]
+            if self.is_job(vmid):
+                continue
+            environment = record.get('observed', {}).get('image', {}).get('defaults', {}).get('Env') or []
+            database = service.get('kind') in database_kinds or any(
+                isinstance(value, str) and value.startswith(markers) for value in environment)
+            unknown_backend = vmid != self.plan['primary_vmid'] and not service.get('frontend')
+            external = any(mount.get('type') == 'host-bind' and not mount.get('read_only')
+                           for mount in record['deployment'].get('mounts', []))
+            if external and (database or unknown_backend):
+                raise ValueError(translate('Migration jobs cannot safely restore database or unclassified backend data from host directories'))
+
    def preflight(self):
        self.validate(self.plan)
+        self._migration_backups()
        ha = json.loads(instances.command('pvesh', 'get', '/cluster/ha/resources', '--output-format', 'json'))
        if not isinstance(ha, list) or any(r.get('sid') == 'ct:%s' % vmid for r in ha for vmid in self.records):
            raise ValueError(translate('High availability resources are not supported for stacks'))
@@ -220,7 +254,7 @@ class NativeAdapter:
            member_tx.preflight(record, record, config, coordinated=True)
            member_tx.freeze_host_sources(record, record, self.acknowledge)
            check = self.services[vmid].get('healthcheck', {})
-            if check.get('type') not in ('exec', 'http', 'running'):
+            if check.get('type') not in ('exec', 'http', 'running', 'tcp', 'completed'):
                raise ValueError(translate('A member has no reproducible service check'))
            if check['type'] == 'exec' and not check.get('argv'):
                raise ValueError(translate('Empty exec service check'))
@@ -231,6 +265,16 @@ class NativeAdapter:
                raise ValueError(translate('HTTP service check without a saved URL'))
            if check['type'] == 'http' and not check['url'].startswith(('http://', 'https://')):
                raise ValueError(translate('Invalid service check URL'))
+            if check['type'] == 'tcp':
+                tcp_target(check)
+            if check['type'] == 'completed':
+                oci_stack_plan.validate_job_check(check, record)
+                expected = compose_jobs.job_spec_from_record(record, config.decode(), check['timeout_seconds'])
+                if (check['job'] != expected or record['deployment'].get('onboot')
+                        or record['deployment'].get('watchdog')
+                        or record['deployment'].get('execution_mode') != 'one-shot'
+                        or parse_config(config).get('onboot', '0') != '0' or self.is_running(vmid)):
+                    raise ValueError(translate('Invalid service check arguments'))
            if not 0 < int(check.get('timeout_seconds', 120)) <= 3600:
                raise ValueError(translate('Invalid service check timeout'))
        primary_id = self.plan['primary_vmid']
@@ -382,12 +426,83 @@ class NativeAdapter:
            progress=f"{translate('Recreating') if operation == 'recreate' else translate('Updating')} "
                     f"{self.describe(vmid)}:")

+    def is_job(self, vmid):
+        return self.services[vmid].get('healthcheck', {}).get('type') == 'completed'
+
+    def _job_contract(self, rollback=False):
+        """Rebind jobs to staged/restored images and journal the hook transition."""
+        if not any(self.is_job(vmid) for vmid in self.services):
+            return
+        primary = self.plan['primary_vmid']
+        path = Path('/etc/pve/priv/proxmenux-stack-%s.json' % primary)
+        info = path.lstat()
+        if not stat.S_ISREG(info.st_mode) or info.st_uid != 0 or info.st_mode & 0o077:
+            raise ValueError(translate('Unsafe dependency hook contract'))
+        current = json.loads(path.read_text())
+        operation = self.state()['id']
+        marker = self.journal.with_name('completed-job-contract.json')
+        if marker.is_symlink():
+            raise ValueError(translate('Unsafe dependency hook contract'))
+        state = {'schema': 1, 'operation_id': operation}
+        if marker.exists():
+            info = marker.lstat()
+            if not stat.S_ISREG(info.st_mode) or info.st_uid != os.geteuid() or info.st_mode & 0o077:
+                raise ValueError(translate('Unsafe dependency hook contract'))
+            state = json.loads(marker.read_text())
+            if state.get('schema') != 1 or state.get('operation_id') != operation:
+                raise ValueError(translate('The dependency hook and the stack recipe differ'))
+        original = next(member for member in self.plan['members'] if member['vmid'] == primary)
+        original_services = original['stack']['deployment']['services']
+        expected = state.get('job_hook_contract')
+        intent = state.get('job_hook_contract_intent')
+        if expected is None:
+            expected = {'schema': 1, 'stack': original['stack']['deployment'].get('stack_name', '')}
+            expected['dependencies'] = [{'vmid': s['vmid'], 'label': s['name'], 'healthcheck': s['healthcheck']}
+                                        for s in original_services if s['vmid'] != primary and not s.get('deferred_setup')]
+        accepted = [expected]
+        if intent is not None:
+            accepted.extend([intent['before'], intent['after']])
+        if current not in accepted:
+            raise ValueError(translate('The dependency hook and the stack recipe differ'))
+        if rollback:
+            self.services = {s['vmid']: copy.deepcopy(s) for s in original_services}
+        for vmid, service in self.services.items():
+            if not self.is_job(vmid):
+                continue
+            record = instances.read(self.root, vmid)
+            config = member_tx.owned(vmid, record['installation_id']).decode()
+            timeout = service['healthcheck'].get('timeout_seconds', 300)
+            service['healthcheck'] = {'type': 'completed', 'timeout_seconds': timeout,
+                                     'job': compose_jobs.job_spec_from_record(record, config, timeout)}
+        rebuilt = copy.deepcopy(current)
+        rebuilt['dependencies'] = [{'vmid': s['vmid'], 'label': s['name'], 'healthcheck': s['healthcheck']}
+                                   for s in self.services.values() if s['vmid'] != primary and not s.get('deferred_setup')]
+        if current != rebuilt or intent is not None:
+            state['job_hook_contract_intent'] = {'before': current, 'after': rebuilt}
+            instances.write(marker, state)
+            instances.write(path, rebuilt)
+            state['job_hook_contract'] = rebuilt
+            state.pop('job_hook_contract_intent', None)
+            instances.write(marker, state)
+
+    def _ensure_job(self, vmid, force=False):
+        self._job_contract()
+        return compose_jobs.ensure_completed(vmid, self.services[vmid]['healthcheck']['job'], force=force)
+
    def start(self, vmid):
        self.validate(self.plan)
+        if self.is_job(vmid):
+            self._ensure_job(vmid)
+            return
+        if vmid == self.plan['primary_vmid']:
+            self._job_contract()
        if not self.is_running(vmid):
            member_tx.run('pct', 'start', str(vmid))

    def healthcheck(self, vmid):
+        if self.is_job(vmid):
+            self._ensure_job(vmid)
+            return
        check = self.services[vmid]['healthcheck']
        timeout = int(check.get('timeout_seconds', 120))
        if not 0 < timeout <= 3600:
@@ -401,6 +516,9 @@ class NativeAdapter:
                    member_tx.run('pct', 'exec', str(vmid), '--', *check['argv'])
                elif check['type'] == 'http':
                    member_tx.run('curl', '-fsS', '--noproxy', '*', '--max-time', '5', check['url'])
+                elif check['type'] == 'tcp':
+                    with socket.create_connection(tcp_target(check), timeout=5):
+                        pass
                member_tx.gpu_devices.validate_runtime(vmid, self.records[vmid]['deployment'])
                deployment = self.records[vmid]['deployment']
                if deployment.get('replay_profile') == {'adapter': 'install_immich_stack.sh', 'role': 'machine-learning'}:
@@ -419,7 +537,7 @@ class NativeAdapter:
                            'assert (any(d.startswith("GPU") for d in ort.capi._pybind_state.get_available_openvino_device_ids()) '
                            'if p=="openvino" else ctypes.CDLL("libcuda.so.1").cuInit(0)==0)', acceleration)
                return
-            except RuntimeError:
+            except (RuntimeError, OSError):
                time.sleep(2)
        raise ValueError(f"{translate('A member did not pass its service check:')} {self.describe(vmid)}")

@@ -441,12 +559,22 @@ class NativeAdapter:
            instances.write(journal, child)

    def restore_running_state(self, running, order):
+        recovering = self.state().get('phase') == 'recovering'
+        has_jobs = any(self.is_job(vmid) for vmid in order)
        for vmid in order:
+            if self.is_job(vmid):
+                if recovering and running.get(str(self.plan['primary_vmid'])):
+                    self._ensure_job(vmid, force=True)
+                continue
            if running[str(vmid)]:
                self.start(vmid)
-        for vmid in order:
-            if running[str(vmid)]:
-                self.healthcheck(vmid)
+                if has_jobs:
+                    # A rollback migration may require the restored database to be ready.
+                    self.healthcheck(vmid)
+        if not has_jobs:
+            for vmid in order:
+                if running[str(vmid)]:
+                    self.healthcheck(vmid)
        # Starting the primary can start dependencies through its native hook.
        for vmid in reversed(order):
            if not running[str(vmid)]:
@@ -465,7 +593,7 @@ class NativeAdapter:
            # Keep child journals available even if publication is interrupted.
            links = self.journal.parent / ('member-%s.json' % vmid)
            instances.write(links, {'journal': str(journal)})
-            member_tx.run('pct', 'set', str(vmid), '--onboot', '1' if original['deployment']['onboot'] else '0')
+            member_tx.run('pct', 'set', str(vmid), '--onboot', '1' if original['deployment']['onboot'] and not self.is_job(vmid) else '0')
            after = member_tx.owned(vmid, original['installation_id'])
            if ([line for line in before.splitlines() if not line.startswith(b'onboot: ')]
                    != [line for line in after.splitlines() if not line.startswith(b'onboot: ')]):
@@ -476,6 +604,7 @@ class NativeAdapter:
        primary_id = self.plan['primary_vmid']
        primary = instances.read(self.root, primary_id)
        primary['stack']['members'] = []
+        primary['stack']['deployment']['services'] = copy.deepcopy(list(self.services.values()))
        for vmid in self.plan['start_order']:
            record = instances.read(self.root, vmid)
            record.pop('stack', None)
@@ -555,8 +684,12 @@ class NativeAdapter:
            if any(self.plan.get(flag) for flag in ('nextcloud_replay', 'paperless_replay', 'tandoor_replay', 'immich_replay')):
                record['deployment']['native_config'] = record['observed']['config']
                record['deployment']['member_replay_projection'] = replay.normalize(record)
+            if self.is_job(vmid) and self.state().get('replacement_intent'):
+                # Restored database contents invalidate an earlier completion proof too.
+                record['recovered_at'] = instances.now()
            record['pending_stack_transaction'] = str(self.journal)
            instances.write(instances.location(self.root, vmid), record)
+        self._job_contract(rollback=True)
        primary_id = plan['primary_vmid']
        primary = instances.read(self.root, primary_id)
        snapshots = []
@@ -566,6 +699,7 @@ class NativeAdapter:
            snapshot.pop('pending_stack_transaction', None)
            snapshots.append(snapshot)
        primary['stack']['members'] = snapshots
+        primary['stack']['deployment']['services'] = copy.deepcopy(list(self.services.values()))
        instances.write(instances.location(self.root, primary_id), primary)

    def finalize(self, state):
@@ -1,9 +1,37 @@
 """Read-only validation of a coordinated stack operation, before host changes."""
 import copy
+import re
+import uuid

 from oci_ui import translate


+def validate_job_check(check, record=None):
+    """Validate a completion proof request without treating STOPPED as success."""
+    failure = translate('Invalid service check arguments')
+    if not isinstance(check, dict) or check.get('type') != 'completed':
+        raise ValueError(failure)
+    job = check.get('job')
+    if not isinstance(job, dict):
+        raise ValueError(failure)
+    try:
+        identity = str(uuid.UUID(job['installation_id']))
+    except (KeyError, ValueError, TypeError, AttributeError) as error:
+        raise ValueError(failure) from error
+    timeout = job.get('timeout_seconds')
+    if (type(timeout) is not int or not 30 <= timeout <= 3600
+            or type(check.get('timeout_seconds')) is not int or check['timeout_seconds'] != timeout
+            or not isinstance(job.get('image_digest'), str)
+            or not re.fullmatch(r'sha256:[0-9a-f]{64}', job['image_digest'])
+            or not isinstance(job.get('expected_config_fingerprint'), str)
+            or not re.fullmatch(r'[0-9a-f]{64}', job['expected_config_fingerprint'])):
+        raise ValueError(failure)
+    if record is not None and (record.get('installation_id') != identity
+            or record.get('observed', {}).get('resolved_registry_digest') != job['image_digest']):
+        raise ValueError(failure)
+    return job
+
+
 def build(primary, records, inventory, operation):
    """Inventory maps VMIDs to installation UUIDs, including unrelated guests.

@@ -48,6 +76,13 @@ def build(primary, records, inventory, operation):
            missing.append(vmid)
        if current.get('deployment', {}).get('rootfs_adaptation_replay_required'):
            blockers.append({'vmid': vmid, 'reason': 'dedicated-adapter-replay-required'})
+        service = next(service for service in services if service['vmid'] == vmid)
+        if service.get('healthcheck', {}).get('type') == 'completed':
+            validate_job_check(service['healthcheck'], current)
+            if (vmid == primary['vmid'] or current.get('deployment', {}).get('onboot')
+                    or current.get('deployment', {}).get('watchdog')
+                    or current.get('deployment', {}).get('execution_mode') != 'one-shot'):
+                raise ValueError(translate('Invalid service check arguments'))
        members.append(copy.deepcopy(current))
    if missing and operation == 'update':
        raise ValueError(translate('Stack members are missing; recreate them after verifying their volumes'))
@@ -124,7 +124,8 @@ def watched(root):
        try:
            vmid = int(path.parent.name)
            record = instances.read(root, vmid)
-            if record.get('status') != 'installed' or record.get('deployment', {}).get('watchdog') is not True:
+            if (record.get('status') != 'installed' or record.get('deployment', {}).get('watchdog') is not True
+                    or record.get('deployment', {}).get('execution_mode') == 'one-shot'):
                continue
            if record.get('pending_stack_transaction') or record.get('pending_transaction'):
                continue
@@ -326,7 +327,8 @@ def set_watchdog(root, vmid, enabled):
        vmids = application(root, vmid)
        for member in vmids:
            record = instances.read(root, member)
-            record.setdefault('deployment', {})['watchdog'] = bool(enabled)
+            deployment = record.setdefault('deployment', {})
+            deployment['watchdog'] = bool(enabled) and deployment.get('execution_mode') != 'one-shot'
            instances.write(instances.location(root, member), record)
            oci_carried_record.carry(root, member, mount_stopped=False)
    if enabled:
@@ -4,6 +4,7 @@ set -Eeuo pipefail
 PATH=/usr/sbin:/usr/bin:/sbin:/bin

 HOOK_NAME=proxmenux-stack-dependencies.sh
+COMPOSE_JOB_HELPER=/var/lib/vz/snippets/proxmenux-compose-jobs.py
 # Longest wait for a dependency that is being backed up, and for any other lock.
 BACKUP_LOCK_WAIT=${PROXMENUX_STACK_BACKUP_LOCK_WAIT:-14400}
 OTHER_LOCK_WAIT=${PROXMENUX_STACK_OTHER_LOCK_WAIT:-300}
@@ -12,6 +13,8 @@ RESTART_GRACE=${PROXMENUX_STACK_RESTART_GRACE:-20}
 POLL=${PROXMENUX_STACK_POLL:-5}
 # Earlier published versions of this script, replaced on update.
 PREVIOUS_RELEASES=(
+  4bc477bcc35e229d1c964c9372838f0973cbe6f1e3e41f0fb486e9b724247310
+  243b0dca81b1b155ca169a4d0e11cdb8ca55974c65ede6a4a9be8f4d536a141e
  c28b69acd7cc8e66d3aec3e8b2ec86bda0e4f91193ec78549f1965f2c859c0d1
  02c50ef17f2411ef22f4d382c51b34a7a0974a446d4a155ecb63fb1e805a7c59
 )
@@ -58,11 +61,40 @@ install_hook() {
    (all(.dependencies[];
      (.vmid | type == "number") and
      (.label | type == "string") and
-      (.healthcheck.type | IN("exec", "http", "running")) and
-      (.healthcheck.timeout_seconds | type == "number")
+      (.healthcheck.type | IN("exec", "http", "running", "tcp", "completed")) and
+      (.healthcheck.timeout_seconds | type == "number") and
+      (if .healthcheck.type == "tcp" then
+        (.healthcheck.host | type == "string") and
+        (.healthcheck.host | test("^([0-9]{1,3}\\.){3}[0-9]{1,3}$")) and
+        (.healthcheck.host | split(".") | all(.[]; tonumber < 256)) and
+        (.healthcheck.port | type == "number") and
+        (.healthcheck.port > 0 and .healthcheck.port < 65536 and .healthcheck.port == (.healthcheck.port | floor))
+       elif .healthcheck.type == "completed" then
+        (.healthcheck.job | type == "object") and
+        (.healthcheck.job.installation_id | type == "string") and
+        (.healthcheck.job.installation_id | test("^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$")) and
+        (.healthcheck.job.image_digest | type == "string") and
+        (.healthcheck.job.image_digest | test("^sha256:[0-9a-f]{64}$")) and
+        (.healthcheck.job.expected_config_fingerprint | type == "string") and
+        (.healthcheck.job.expected_config_fingerprint | test("^[0-9a-f]{64}$")) and
+        (.healthcheck.job.timeout_seconds == .healthcheck.timeout_seconds) and
+        (.healthcheck.timeout_seconds >= 30 and .healthcheck.timeout_seconds <= 3600 and .healthcheck.timeout_seconds == (.healthcheck.timeout_seconds | floor))
+       else true end)
    ))
  ' "$source_config" >/dev/null || die "Invalid dependency contract"

+  if jq -e 'any(.dependencies[]; .healthcheck.type == "completed")' "$source_config" >/dev/null; then
+    local helper_source helper_temporary
+    helper_source="$(dirname -- "$0")/oci_compose_jobs.py"
+    [[ -f $helper_source && ! -L $helper_source ]] || die "Native OCI job helper is missing"
+    install -d -m 0755 "$(dirname -- "$COMPOSE_JOB_HELPER")"
+    helper_temporary=$(mktemp "${COMPOSE_JOB_HELPER}.XXXXXX")
+    if ! install -m 0755 "$helper_source" "$helper_temporary" || ! mv -f -- "$helper_temporary" "$COMPOSE_JOB_HELPER"; then
+      rm -f -- "$helper_temporary"
+      die "Could not install the persistent native OCI job helper"
+    fi
+  fi
+
  storage=$(find_snippet_storage)
  hook_volume="${storage}:snippets/${HOOK_NAME}"
  hook_path=$(pvesm path "$hook_volume")
@@ -144,6 +176,20 @@ dependency_is_healthy() {
      url=$(jq -r '.url' <<<"$healthcheck")
      curl -fsS --max-time 3 "$url" >/dev/null 2>&1
      ;;
+    tcp)
+      local host port
+      host=$(jq -r '.host' <<<"$healthcheck")
+      port=$(jq -r '.port' <<<"$healthcheck")
+      python3 - "$host" "$port" <<'PYTCP' >/dev/null 2>&1
+import ipaddress, socket, sys
+host = str(ipaddress.IPv4Address(sys.argv[1]))
+port = int(sys.argv[2])
+if not 0 < port < 65536:
+    raise ValueError('Invalid TCP port')
+with socket.create_connection((host, port), timeout=3):
+    pass
+PYTCP
+      ;;
    *) return 1 ;;
  esac
 }
@@ -167,6 +213,20 @@ start_dependencies() {
    pct config "$id" >/dev/null 2>&1 \
      || die "Dependency $label (CT $id) does not exist"

+    if [[ $(jq -r '.type' <<<"$healthcheck") == completed ]]; then
+      [[ -f $COMPOSE_JOB_HELPER && ! -L $COMPOSE_JOB_HELPER ]] || die "Native OCI job helper is missing"
+      local helper_owner helper_mode
+      helper_owner=$(stat -c '%u' "$COMPOSE_JOB_HELPER")
+      helper_mode=$(stat -c '%a' "$COMPOSE_JOB_HELPER")
+      [[ $helper_owner == 0 && $helper_mode =~ ^[0-7]{3,4}$ ]] && (( (8#$helper_mode & 8#022) == 0 )) \
+        || die "Native OCI job helper is not owned safely by root"
+      printf 'Waiting for completed job %s (CT %s)...\n' "$label" "$id"
+      python3 -I "$COMPOSE_JOB_HELPER" ensure "$id" --spec-json "$(jq -c '.job' <<<"$healthcheck")" >/dev/null 9>&- \
+        || die "Job $label (CT $id) did not complete successfully"
+      printf 'Dependency %s (CT %s): ready.\n' "$label" "$id"
+      continue
+    fi
+
    is_running "$id" || wait_for_release "$id" "$label"
    if ! is_running "$id"; then
      printf 'Starting dependency %s (CT %s)...\n' "$label" "$id"