baiyue-firstleg/scripts/order-submit-dependency-fault-runtime.py

"""Bounded remote TEST actions; injected by the JS executor, never invoked at import.

Only Order is recreated. Every Docker mutation is journaled before sending; an
uncertain result blocks further mutations until explicit read-only reconciliation.
"""
import ast
import base64
import copy
import hashlib
import http.client
import json
import os
import re
import secrets
import signal
import socket
import subprocess
import sys
import time
import urllib.parse
import urllib.request
import urllib.error
from pathlib import Path

ORDER = 'firstleg-test-order'
ENTRYPOINT = ['sh', '-c', 'exec java $JAVA_OPTS org.springframework.boot.loader.launch.JarLauncher']
PROPERTIES = {'IAM': 'baiyue.clients.iam.url', 'FINANCE': 'baiyue.clients.finance.url',
              'WAREHOUSE': 'baiyue.clients.warehouse.url'}
UPSTREAM_CONTAINERS = {'IAM': 'firstleg-test-iam', 'FINANCE': 'firstleg-test-finance',
                       'WAREHOUSE': 'firstleg-test-warehouse'}


class NoRedirect(urllib.request.HTTPRedirectHandler):
    def redirect_request(self, *_args, **_kwargs): return None


def http_open(request, timeout):
    return urllib.request.build_opener(NoRedirect()).open(request, timeout=timeout)


def common_tools(source):
    """Reuse only Docker read/config reconstruction primitives, never event/SQL functions."""
    tree = ast.parse(source)
    allowed = {'DockerConnection', 'docker', 'inspect', 'env_of', 'require_test', 'recreation', 'fingerprint'}
    nodes = [node for node in tree.body if isinstance(node, (ast.Import, ast.ImportFrom))
             or isinstance(node, (ast.FunctionDef, ast.ClassDef)) and node.name in allowed]
    namespace = {'__name__': 'order_dependency_primitives'}
    exec(compile(ast.Module(body=nodes, type_ignores=[]), '<reviewed-docker-primitives>', 'exec'), namespace)
    return {name: namespace[name] for name in allowed}


def digest(value):
    return hashlib.sha256(json.dumps(value, sort_keys=True, separators=(',', ':')).encode()).hexdigest()


def private_write(path, value):
    temporary = path.with_name(path.name + '.next')
    fd = os.open(temporary, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
    with os.fdopen(fd, 'w') as stream:
        json.dump(value, stream); stream.flush(); os.fsync(stream.fileno())
    os.replace(temporary, path)
    parent = os.open(path.parent, os.O_RDONLY)
    try: os.fsync(parent)
    finally: os.close(parent)


def validate_scope(scope):
    if set(scope) != {'marker', 'customerId', 'receivingWarehouseId', 'orderIds', 'planSha256'} or not re.fullmatch(r'[A-Za-z0-9_-]{1,80}', scope['marker']):
        raise RuntimeError('INVALID_SCOPE')
    if not re.fullmatch(r'[1-9][0-9]{0,19}', scope['customerId']) or set(scope['orderIds']) != {'merchant', 'platform'}:
        raise RuntimeError('INVALID_SCOPE')
    if not re.fullmatch(r'[1-9][0-9]{0,19}', scope['receivingWarehouseId']):
        raise RuntimeError('INVALID_RECEIVING_WAREHOUSE')
    if any(not re.fullmatch(r'[1-9][0-9]{0,19}', value) for value in scope['orderIds'].values()) or len(set(scope['orderIds'].values())) != 2:
        raise RuntimeError('TWO_DISTINCT_ORDERS_REQUIRED')
    if not re.fullmatch(r'[0-9a-f]{64}', scope['planSha256']): raise RuntimeError('INVALID_PLAN_HASH')


class Runtime:
    def __init__(self, scope, tools, base=Path('/tmp')):
        validate_scope(scope)
        self.scope, self.tools = scope, tools
        self.directory = base / ('chongchong-order-dependency-' + scope['marker'])
        self.path = self.directory / 'state.json'
        self.lock = base / 'chongchong-order-dependency.lock'

    def save(self, state): private_write(self.path, state)

    def load(self):
        if self.directory.is_symlink() or self.directory.stat().st_uid != os.getuid() or self.directory.stat().st_mode & 0o077:
            raise RuntimeError('UNSAFE_STATE_DIRECTORY')
        if self.path.is_symlink() or self.path.stat().st_mode & 0o077: raise RuntimeError('UNSAFE_STATE_FILE')
        state = json.loads(self.path.read_text())
        if state['scope'] != self.scope: raise RuntimeError('SCOPE_CHANGED')
        if (state.get('mutation') or {}).get('step', '').endswith('_INTENT'):
            state['uncertain'] = True  # Includes process death between HTTP completion and journal commit.
        return state

    def container(self):
        value = self.tools['inspect'](ORDER)
        self.tools['require_test'](value, ORDER)
        if self.tools['env_of'](value).get('DEPLOY_ENV') != 'test': raise RuntimeError('NOT_TEST_ENV')
        return value

    def control(self, state, mode=None, ttl=0):
        pid = state.get('proxyPid')
        if not isinstance(pid, int) or pid <= 1: raise RuntimeError('PROXY_PID_NOT_READY')
        process = Path('/proc') / str(pid)
        if str(self.directory / 'proxy.py').encode() not in (process / 'cmdline').read_bytes().split(b'\0'):
            raise RuntimeError('PROXY_PROCESS_OWNERSHIP_MISMATCH')
        socket_inodes = set()
        for descriptor in (process / 'fd').iterdir():
            try: target = os.readlink(descriptor)
            except FileNotFoundError: continue
            match = re.fullmatch(r'socket:\[(\d+)\]', target)
            if match: socket_inodes.add(match.group(1))
        owned_ports = set()
        for line in Path('/proc/net/tcp').read_text().splitlines()[1:]:
            values = line.split()
            if len(values) > 9 and values[3] == '0A' and values[9] in socket_inodes:
                owned_ports.add(int(values[1].split(':')[1], 16))
        if not {state['proxyPort'], state['controlPort']}.issubset(owned_ports): raise RuntimeError('PROXY_LISTENERS_NOT_OWNED')
        req = urllib.request.Request('http://127.0.0.1:' + str(state['controlPort']) + ('/mode' if mode else '/status'),
            None if mode is None else json.dumps({'mode': mode, 'ttl': ttl}).encode(),
            {'X-Order-Fault-Control': state['controlToken'], 'Content-Type': 'application/json'})
        with http_open(req, timeout=5) as response: return json.load(response)

    def probes(self, state):
        """Real authenticated reads, not manufactured business responses. No body/secret is returned."""
        customer = self.scope['customerId']; results = {}
        targets = {'IAM': '/internal/iam/tenants/' + customer + '/order-eligibility',
                   'FINANCE': '/internal/finance/v1/wallet/accounts/' + customer,
                   'WAREHOUSE': '/internal/warehouses/0/' + self.scope['receivingWarehouseId'] + '/active'}
        for dependency, path in targets.items():
            req = urllib.request.Request(state['upstreams'][dependency] + path,
                headers={'X-Internal-Service-Token': state['internalToken']})
            with http_open(req, timeout=8) as response:
                value = json.load(response)
                if response.status != 200 or str(value.get('code')) != '200': raise RuntimeError('UPSTREAM_PROBE_FAILED')
            data = value.get('data') or {}
            if dependency == 'IAM' and data.get('eligible') is not True: raise RuntimeError('CUSTOMER_NOT_ELIGIBLE')
            if dependency == 'FINANCE' and (data.get('actualBalance') is None or float(data['actualBalance']) < 0):
                raise RuntimeError('CUSTOMER_BALANCE_NOT_READY')
            if dependency == 'WAREHOUSE':
                warehouse = data.get('warehouse') or {}
                if (str(warehouse.get('id')) != self.scope['receivingWarehouseId']
                        or str(warehouse.get('tenantId')) != '0' or warehouse.get('status') != 'ACTIVE'
                        or warehouse.get('warehouseType') != 'DOMESTIC' or not warehouse.get('warehouseName')):
                    raise RuntimeError('RECEIVING_WAREHOUSE_NOT_READY')
            results[dependency] = True
        return results

    def config_fingerprint(self, container, routing):
        """Read the two actual Nacos imports using the current v3 Admin endpoints.
        No endpoint guessing/fallback and no Nacos write operation is supported.
        """
        env = self.tools['env_of'](container); config = routing.get('nacos', {})
        base = urllib.parse.urlsplit(config.get('baseUrl', ''))
        configured = urllib.parse.urlsplit(env.get('NACOS_SERVER_ADDR', '') if '://' in env.get('NACOS_SERVER_ADDR', '')
                                         else 'http://' + env.get('NACOS_SERVER_ADDR', ''))
        allowed_hosts = {configured.hostname, '127.0.0.1', 'localhost'} | {
            network.get('Gateway') for network in container['NetworkSettings']['Networks'].values()}
        if (base.scheme not in ('http', 'https') or base.hostname not in allowed_hosts or base.port != configured.port
                or base.path not in ('', '/') or base.username or base.password or base.query or base.fragment
                or config.get('loginPath') != '/nacos/v3/auth/user/login'
                or config.get('configPath') != '/nacos/v3/admin/cs/config'):
            raise RuntimeError('EXPLICIT_NACOS_READ_ENDPOINTS_REQUIRED')
        if any(not env.get(key) for key in ['NACOS_USERNAME', 'NACOS_PASSWORD', 'NACOS_GROUP', 'NACOS_NAMESPACE']):
            raise RuntimeError('NACOS_READ_IDENTITY_UNAVAILABLE')
        origin = config['baseUrl'].rstrip('/')
        request = urllib.request.Request(origin + config['loginPath'], urllib.parse.urlencode({
            'username': env['NACOS_USERNAME'], 'password': env['NACOS_PASSWORD']}).encode(),
            {'Content-Type': 'application/x-www-form-urlencoded'})
        with http_open(request, timeout=8) as response: token = json.load(response).get('accessToken')
        if not isinstance(token, str) or not token: raise RuntimeError('NACOS_READ_LOGIN_FAILED')
        fingerprints = {}
        for data_id in ('baiyue-common.yml', 'order-service.yml'):
            query = urllib.parse.urlencode({'dataId': data_id, 'groupName': env['NACOS_GROUP'], 'namespaceId': env['NACOS_NAMESPACE']})
            request = urllib.request.Request(origin + config['configPath'] + '?' + query,
                headers={'Authorization': 'Bearer ' + token})
            with http_open(request, timeout=8) as response:
                if response.status != 200: raise RuntimeError('NACOS_SOURCE_READ_FAILED')
                value = json.load(response)
            data = value.get('data')
            if value.get('code') != 0 or not isinstance(data, dict): raise RuntimeError('NACOS_SOURCE_RESPONSE_INVALID')
            if (data.get('dataId') != data_id or data.get('groupName') != env['NACOS_GROUP']
                    or data.get('namespaceId') != env['NACOS_NAMESPACE']): raise RuntimeError('NACOS_SOURCE_IDENTITY_MISMATCH')
            content = data.get('content')
            if not isinstance(content, str) or not content.strip() or content.lstrip().startswith(('{', '<')):
                raise RuntimeError('NACOS_SOURCE_NOT_YAML')
            content = content.encode('utf-8')
            fingerprints[data_id] = hashlib.sha256(content).hexdigest()
        return digest({'imports': fingerprints, 'namespace': env['NACOS_NAMESPACE'], 'group': env['NACOS_GROUP']})

    def prepare(self, config):
        if self.directory.exists(): raise RuntimeError('STATE_EXISTS_USE_STATUS_OR_RESTORE')
        original = self.container()
        if original['Image'] != config.get('expectedImageId') or not re.fullmatch(r'sha256:[0-9a-f]{64}', original['Image']):
            raise RuntimeError('ORIGINAL_IMAGE_MISMATCH')
        if original['State'].get('Health', {}).get('Status') != 'healthy': raise RuntimeError('ORDER_NOT_HEALTHY')
        if original['Config']['Entrypoint'] != ENTRYPOINT or original['Config'].get('Cmd') not in (None, []):
            raise RuntimeError('UNREVIEWED_JAVA_ENTRYPOINT')
        if 'host.docker.internal:host-gateway' not in (original['HostConfig'].get('ExtraHosts') or []):
            raise RuntimeError('REVIEWED_PROXY_HOST_ROUTE_REQUIRED')
        env = self.tools['env_of'](original)
        token = env.get('INTERNAL_SERVICE_TOKEN', '')
        if len(token) < 16: raise RuntimeError('INTERNAL_TOKEN_UNAVAILABLE')
        routing = config.get('routing', {})
        if routing.get('configurationReviewed') is not True:
            raise RuntimeError('UPSTREAM_CONFIGURATION_REVIEW_REQUIRED')
        source_digest = self.config_fingerprint(original, routing)
        if routing.get('sourceSha256') != source_digest: raise RuntimeError('REVIEWED_NACOS_SOURCE_DIGEST_CHANGED')
        upstreams = routing.get('upstreams', {})
        if set(upstreams) != set(PROPERTIES): raise RuntimeError('EXPLICIT_UPSTREAMS_REQUIRED')
        for dependency, upstream in upstreams.items():
            url = urllib.parse.urlsplit(upstream)
            target = self.tools['inspect'](UPSTREAM_CONTAINERS[dependency]); self.tools['require_test'](target, UPSTREAM_CONTAINERS[dependency])
            target_env = self.tools['env_of'](target)
            ips = {network['IPAddress'] for name, network in target['NetworkSettings']['Networks'].items()
                   if name in original['NetworkSettings']['Networks']}
            if (url.scheme != 'http' or url.username or url.password or url.path not in ('', '/') or url.query or url.fragment
                    or url.hostname not in ips or str(url.port) != target_env.get('SERVER_PORT')
                    or target_env.get('DEPLOY_ENV') != 'test' or target['State'].get('Health', {}).get('Status') != 'healthy'):
                raise RuntimeError('UPSTREAM_NOT_VERIFIED_TEST_CONTAINER')
        lock_fd = os.open(self.lock, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
        with os.fdopen(lock_fd, 'w') as stream: stream.write(self.scope['marker']); stream.flush(); os.fsync(stream.fileno())
        self.directory.mkdir(mode=0o700)
        sockets = [socket.socket(), socket.socket()]
        for sock in sockets: sock.bind(('127.0.0.1', 0))
        proxy_port, control_port = [sock.getsockname()[1] for sock in sockets]
        for sock in sockets: sock.close()
        state = {'scope': self.scope, 'phase': 'PREPARED', 'originalId': original['Id'],
                 'originalImageId': original['Image'], 'originalFingerprint': self.tools['fingerprint'](original),
                 'originalBody': self.tools['recreation'](original, env, full=True), 'internalToken': token,
                 'proxyPort': proxy_port, 'controlPort': control_port, 'controlToken': secrets.token_urlsafe(32),
                 'upstreams': {k: v.rstrip('/') for k, v in upstreams.items()}, 'routingSourceSha256': source_digest, 'routing': routing,
                 'orderIps': [x['IPAddress'] for x in original['NetworkSettings']['Networks'].values()], 'mutation': None, 'proxyLaunchIntent': False}
        self.save(state)  # Recovery info exists before any process or container mutation.
        self.probes(state)
        proxy = self.directory / 'proxy.py'; proxy.write_bytes(base64.b64decode(config['proxySource'])); os.chmod(proxy, 0o600)
        private_write(self.directory / 'proxy-config.json', {k: state[k] for k in
            ['internalToken', 'controlToken', 'proxyPort', 'controlPort', 'upstreams', 'orderIps']} | {'customerId': self.scope['customerId'],
             'receivingWarehouseId': self.scope['receivingWarehouseId'],
             'requestIds': ['dep-' + self.scope['marker'] + '-' + dependency + '-' + role
                            for dependency in PROPERTIES for role in ('merchant', 'platform')]})
        state['phase'] = 'PROXY_START_INTENT'; state['proxyLaunchIntent'] = True; self.save(state)
        with open(self.directory / 'proxy-output.log', 'ab') as output:
            os.chmod(self.directory / 'proxy-output.log', 0o600)
            process = subprocess.Popen(['python3', str(proxy), '--config', str(self.directory / 'proxy-config.json')],
                stdout=output, stderr=output, start_new_session=True)
        state['proxyPid'] = process.pid; state['phase'] = 'PROXY_STARTED'; self.save(state)
        return self.summary(state)

    def summary(self, state):
        return {'scope': self.scope, 'phase': state['phase'], 'originalImageId': state['originalImageId'],
                'originalFingerprint': state['originalFingerprint'], 'uncertain': bool(state.get('uncertain')),
                'mutationStep': (state.get('mutation') or {}).get('step'), 'restored': state.get('restored', False)}

    def mutation(self, state, step, method, path, body=None):
        if state.get('uncertain'): raise RuntimeError('UNCERTAIN_MUTATION_REQUIRES_RECONCILE')
        state['mutation'] = {'step': step, 'path': path, 'bodySha256': digest(body), 'intentAt': time.time()}; self.save(state)
        try:
            result = self.tools['docker'](method, path, body)
            if step == 'CREATE_INTENT' and (not isinstance(result, dict) or not re.fullmatch(r'[A-Za-z0-9_-]{1,128}', result.get('Id', ''))):
                raise RuntimeError('INVALID_CREATE_ACK')
        except Exception:
            state['uncertain'] = True; self.save(state)
            raise RuntimeError('UNCERTAIN_DOCKER_MUTATION')
        state['mutation']['step'] = step.replace('_INTENT', '_DONE')
        if step == 'CREATE_INTENT': state['createdId'] = result['Id']
        self.save(state)
        return result

    def replace(self, state, current, desired, purpose):
        if state.get('uncertain'): raise RuntimeError('UNCERTAIN_MUTATION_REQUIRES_RECONCILE')
        if self.container()['Id'] != current['Id']: raise RuntimeError('CONCURRENT_ORDER_CHANGE')
        image = self.tools['docker']('GET', '/images/' + urllib.parse.quote(desired['Image'], safe='') + '/json')
        if image['Id'] != state['originalImageId']: raise RuntimeError('IMAGE_TAG_MOVED_NO_RECREATE')
        state['purpose'] = purpose; state['desiredBody'] = desired; state['replacingId'] = current['Id']; self.save(state)
        if current['State']['Running']:
            self.mutation(state, 'STOP_INTENT', 'POST', '/containers/' + current['Id'] + '/stop?t=15')
        self.mutation(state, 'DELETE_INTENT', 'DELETE', '/containers/' + current['Id'] + '?v=false')
        created = self.mutation(state, 'CREATE_INTENT', 'POST', '/containers/create?name=' + ORDER, desired)
        self.mutation(state, 'START_INTENT', 'POST', '/containers/' + created['Id'] + '/start')
        state['currentId'] = created['Id']; state['phase'] = purpose; state['mutation'] = None; self.save(state)

    def matches_injected(self, current, state):
        expected = copy.deepcopy(state['originalBody']['Entrypoint'])
        expected[2] += ''.join(' --' + prop + '=http://host.docker.internal:' + str(state['proxyPort']) + '/' + name.lower()
                               for name, prop in PROPERTIES.items())
        restored_view = copy.deepcopy(current)
        restored_view['Config']['Entrypoint'] = state['originalBody']['Entrypoint']
        return (current['Image'] == state['originalImageId'] and current['Config']['Entrypoint'] == expected
                and self.tools['fingerprint'](restored_view) == state['originalFingerprint'])

    def activate(self, state):
        if state['phase'] != 'PROXY_STARTED' or state.get('uncertain'): raise RuntimeError('ACTIVATE_NOT_RETRYABLE')
        if self.control(state)['mode'] != 'PASS': raise RuntimeError('PROXY_NOT_PASS')
        current = self.container()
        if current['Id'] != state['originalId'] or self.tools['fingerprint'](current) != state['originalFingerprint']:
            raise RuntimeError('ORIGINAL_CONFIGURATION_CHANGED')
        desired = copy.deepcopy(state['originalBody'])
        desired['Entrypoint'][2] += ''.join(' --' + prop + '=http://host.docker.internal:' + str(state['proxyPort']) + '/' + name.lower()
                                          for name, prop in PROPERTIES.items())
        self.replace(state, current, desired, 'INJECTED')
        state['injectedFingerprint'] = self.tools['fingerprint'](self.container()); self.save(state)
        return self.summary(state)

    def restore(self, state):
        # PASS is attempted even if Docker state is ambiguous. Never stop the proxy here.
        try: self.control(state, 'PASS')
        except Exception:
            state['passUnconfirmed'] = True; self.save(state)
        if state.get('uncertain'): raise RuntimeError('UNCERTAIN_MUTATION_REQUIRES_RECONCILE')
        if state['phase'] == 'ABSENT_CONFIRMED':
            try: self.tools['inspect'](ORDER)
            except RuntimeError as error:
                if str(error) != 'DOCKER_404': raise
            else: raise RuntimeError('ORDER_REAPPEARED_NO_RECREATE')
            body = copy.deepcopy(state['originalBody'])
            image = self.tools['docker']('GET', '/images/' + urllib.parse.quote(body['Image'], safe='') + '/json')
            if image['Id'] != state['originalImageId']: raise RuntimeError('IMAGE_TAG_MOVED_NO_RECREATE')
            state['purpose'] = 'ORIGINAL_RESTORED'; state['desiredBody'] = body; self.save(state)
            created = self.mutation(state, 'CREATE_INTENT', 'POST', '/containers/create?name=' + ORDER, body)
            self.mutation(state, 'START_INTENT', 'POST', '/containers/' + created['Id'] + '/start')
            state['currentId'] = created['Id']; state['phase'] = 'ORIGINAL_RESTORED'; state['mutation'] = None; self.save(state)
            return self.summary(state)
        current = self.container()
        if self.tools['fingerprint'](current) == state['originalFingerprint'] and current['Image'] == state['originalImageId']:
            if not current['State']['Running']:
                state['purpose'] = 'ORIGINAL_RESTORED'; state['createdId'] = current['Id']; self.save(state)
                self.mutation(state, 'START_INTENT', 'POST', '/containers/' + current['Id'] + '/start')
            state['phase'] = 'ORIGINAL_RESTORED'; self.save(state); return self.summary(state)
        if current['Id'] != state.get('currentId') or not self.matches_injected(current, state):
            raise RuntimeError('UNRECOGNIZED_ORDER_CONFIGURATION')
        self.replace(state, current, copy.deepcopy(state['originalBody']), 'ORIGINAL_RESTORED')
        return self.summary(state)

    def reconcile(self, state):
        # This action is intentionally read-only with respect to Docker. It never resumes deletion/creation.
        if not state.get('uncertain'): return self.summary(state)
        try: current = self.container()
        except RuntimeError as error:
            if str(error) != 'DOCKER_404' or (state.get('mutation') or {}).get('step') != 'DELETE_INTENT':
                raise RuntimeError('MANUAL_DOCKER_RECOVERY_REQUIRED_NO_BLIND_RECREATE')
            try: self.tools['inspect'](state['replacingId'])
            except RuntimeError as by_id:
                if str(by_id) != 'DOCKER_404': raise
            else: raise RuntimeError('ORIGINAL_ID_STILL_PRESENT')
            state['uncertain'] = False; state['phase'] = 'ABSENT_CONFIRMED'; state['mutation'] = None; self.save(state)
            return self.summary(state)
        step = (state.get('mutation') or {}).get('step')
        original = current['Image'] == state['originalImageId'] and self.tools['fingerprint'](current) == state['originalFingerprint']
        injected = self.matches_injected(current, state)
        desired_matches = injected if state.get('purpose') == 'INJECTED' else original
        if step == 'STOP_INTENT' and current['Id'] == state.get('replacingId') and not current['State']['Running'] and (original or injected):
            phase = 'ORIGINAL_RESTORED' if original else 'INJECTED'
        elif step == 'CREATE_INTENT' and current['Id'] != state.get('replacingId') and desired_matches:
            phase = state['purpose']
        elif step == 'START_INTENT' and current['Id'] == state.get('createdId') and current['State']['Running'] and desired_matches:
            phase = state['purpose']
        else:
            raise RuntimeError('MANUAL_DOCKER_RECOVERY_REQUIRED_NO_BLIND_RECREATE')
        state['uncertain'] = False; state['currentId'] = current['Id']; state['phase'] = phase; state['mutation'] = None
        if phase == 'INJECTED': state['injectedFingerprint'] = self.tools['fingerprint'](current)
        self.save(state); return self.summary(state)

    def verify_restored(self, state):
        current = self.container()
        if state.get('uncertain') or current['Image'] != state['originalImageId'] or self.tools['fingerprint'](current) != state['originalFingerprint']:
            raise RuntimeError('RESTORED_CONFIG_OR_IMAGE_MISMATCH')
        if current['State'].get('Health', {}).get('Status') != 'healthy': raise RuntimeError('RESTORED_NOT_HEALTHY')
        if self.config_fingerprint(current, state['routing']) != state['routingSourceSha256']:
            raise RuntimeError('NACOS_ROUTE_CONFIGURATION_CHANGED')
        checks = self.probes(state)
        if checks != {dependency: True for dependency in PROPERTIES}: raise RuntimeError('RESTORED_ROUTE_PROBE_INVALID')
        state['routeReadsVerified'] = checks; self.save(state)
        return {'healthy': True, 'originalFingerprint': state['originalFingerprint'], 'currentFingerprint': self.tools['fingerprint'](current),
                'originalImageId': state['originalImageId'], 'currentImageId': current['Image'], 'upstreamReads': checks,
                'nacosSourcesUnchanged': True}

    def finish(self, state, business):
        self.verify_restored(state)
        if self.lock.is_symlink() or self.lock.read_text() != self.scope['marker']: raise RuntimeError('RUNTIME_LOCK_OWNER_CHANGED')
        if set(business or {}) != {'merchant', 'platform'} or any(
                business[role].get('orderId') != order_id or business[role].get('status') != 'SUBMITTED'
                or business[role].get('sameKeyReplayVerified') is not True for role, order_id in self.scope['orderIds'].items()):
            raise RuntimeError('RESTORED_ORDER_ROUTES_NOT_PROVED')
        pid = state.get('proxyPid')
        if not pid and state.get('proxyLaunchIntent'):
            pid_file = self.directory / 'proxy.pid'
            if pid_file.is_symlink() or not pid_file.exists() or not pid_file.read_text().isdigit(): raise RuntimeError('PROXY_PID_UNCERTAIN')
            pid = int(pid_file.read_text())
        command = Path('/proc/' + str(pid) + '/cmdline') if pid else None
        if command and command.exists():
            if str(self.directory / 'proxy.py').encode() not in command.read_bytes().split(b'\0'):
                raise RuntimeError('PROXY_PROCESS_OWNERSHIP_MISMATCH')
            os.kill(pid, signal.SIGTERM)
            for _ in range(40):
                if not command.exists() or not command.read_bytes(): break
                time.sleep(0.1)
            else: raise RuntimeError('PROXY_NOT_STOPPED')
        state['restored'] = True; state['phase'] = 'FINISHED'
        # Recovery snapshots remain protected until completion; erase secrets only after verified restoration.
        for field in ['originalBody', 'desiredBody', 'internalToken', 'controlToken']: state.pop(field, None)
        self.save(state)
        config = self.directory / 'proxy-config.json'
        if config.exists(): config.unlink()
        self.lock.unlink()
        return {'restored': True, 'healthy': True, 'proxyStopped': True, 'originalImageId': state['originalImageId'],
                'originalFingerprint': state['originalFingerprint'], 'orderRoutesVerified': True}


def execute(config):
    runtime = Runtime(config['scope'], common_tools(base64.b64decode(config['commonSource']).decode()))
    action = config['action']
    if action == 'preflight':
        current = runtime.container()
        return {'originalImageId': current['Image'], 'originalFingerprint': runtime.tools['fingerprint'](current),
                'sourceSha256': runtime.config_fingerprint(current, config['routing']), 'runtimeChanged': False}
    if action == 'prepare': return runtime.prepare(config)
    state = runtime.load()
    if action == 'status':
        result = runtime.summary(state)
        if not state.get('restored'):
            try: result['proxy'] = runtime.control(state)
            except Exception: result['proxy'] = {'unavailable': True}
        return result
    if state.get('restored'): raise RuntimeError('ALREADY_FINISHED_USE_STATUS')
    if action not in ('records', 'health') and (runtime.lock.is_symlink() or runtime.lock.read_text() != runtime.scope['marker']):
        raise RuntimeError('RUNTIME_LOCK_OWNER_CHANGED')
    if action == 'records':
        path = runtime.directory / 'proxy-evidence.jsonl'
        return {'records': [json.loads(line) for line in path.read_text().splitlines()] if path.exists() else []}
    if action == 'activate': return runtime.activate(state)
    if action == 'mode':
        if state['phase'] != 'INJECTED' or state.get('uncertain'): raise RuntimeError('FAULT_WINDOW_NOT_READY')
        return runtime.control(state, config['mode'], config.get('ttl', 0))
    if action == 'health': return {'healthy': runtime.container()['State'].get('Health', {}).get('Status') == 'healthy'}
    if action == 'restore': return runtime.restore(state)
    if action == 'reconcile': return runtime.reconcile(state)
    if action == 'verify-restored': return runtime.verify_restored(state)
    if action == 'finish': return runtime.finish(state, config.get('businessRecovery'))
    raise RuntimeError('UNKNOWN_ACTION')


def main(config):
    # A lost Jenkins response must not allow a second process to race the first Docker mutation.
    import fcntl
    validate_scope(config['scope'])
    path = Path('/tmp') / ('chongchong-order-dependency-' + config['scope']['marker'] + '.action.lock')
    if path.is_symlink(): raise RuntimeError('UNSAFE_ACTION_LOCK')
    fd = os.open(path, os.O_RDWR | os.O_CREAT, 0o600)
    try:
        try: fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB)
        except BlockingIOError: raise RuntimeError('RUNTIME_ACTION_IN_PROGRESS_USE_STATUS_LATER')
        return execute(config)
    finally: os.close(fd)


def safe_error_code(error):
    message = str(error)
    if isinstance(error, urllib.error.HTTPError):
        path = urllib.parse.urlsplit(error.geturl()).path
        stage = 'NACOS_CONFIG' if path == '/nacos/v3/admin/cs/config' else (
            'NACOS_LOGIN' if path == '/nacos/v3/auth/user/login' else 'DEPENDENCY_HTTP')
        return stage + '_' + str(error.code)
    if isinstance(error, urllib.error.URLError):
        return 'DEPENDENCY_TRANSPORT_FAILED'
    return message if re.fullmatch(r'[A-Z_0-9]{1,90}', message) else 'ORDER_DEPENDENCY_ACTION_FAILED'


if __name__ == '__main__':
    try:
        result = main(json.loads(base64.b64decode(sys.argv[1])))
        print('ORDER_DEPENDENCY_JSON=' + json.dumps({'ok': True, 'data': result}))
    except Exception as error:
        print('ORDER_DEPENDENCY_JSON=' + json.dumps({'ok': False, 'error': safe_error_code(error)}))