#!/usr/bin/env python3 """Owned weight chain, preserving previously observed POST read evidence.""" import copy import hashlib import json from pathlib import Path import subprocess import sys import integrate_20261010 as tool from integrate_20261011_batches46_48 import item BASE = tool.AI + 'parallel-ai-user-weight-owned-flow-20261011/' OLD = tool.AI + 'parallel-ai-profile-weight-motion/ABC_BATCH_RESULT.md' PRIVATE = Path('/tmp/cdx-ai-weight-owned-flow-20261011') def read(lane, name): return json.loads((PRIVATE / ('private-' + lane + '-final') / name).read_text()) def semantic_equal(left, right): # Object member order is not the same thing as ordered JSON arrays/binds. if type(left) is not type(right): return False if isinstance(left, dict): return set(left) == set(right) and all(semantic_equal(left[k], right[k]) for k in left) if isinstance(left, list): return len(left) == len(right) and all(semantic_equal(a, b) for a, b in zip(left, right)) return left == right def cross_audit(): actors = {lane: read(lane, 'owned-actor.json') for lane in ['B', 'C']} finals = {lane: read(lane, 'final-capture.json') for lane in ['B', 'C']} pairs = 0 member_order_variances = [] for path in sorted((PRIVATE / 'private-B-final').glob('response-*.json')): label = path.stem[9:] b = json.loads(read('B', path.name)['raw_response_body']) c = copy.deepcopy(json.loads(read('C', path.name)['raw_response_body'])) assert isinstance(b['trace_id'], str) and isinstance(c['trace_id'], str) assert b['trace_id'] and c['trace_id'] c['trace_id'] = b['trace_id'] if label in ['W1_MANUAL', 'W2_GOAL', 'R5_STATUS_EXISTING_GOAL']: assert b['data']['user_id'] == actors['B']['user_id'] assert c['data']['user_id'] == actors['C']['user_id'] c['data']['user_id'] = b['data']['user_id'] if label in ['W1_MANUAL', 'W2_GOAL']: assert b['data']['nick_name'] == actors['B']['marker'] assert c['data']['nick_name'] == actors['C']['marker'] c['data']['nick_name'] = b['data']['nick_name'] if list(b['data']) != list(c['data']): moved = ['withdrawable_balance', 'activity_principal_balance'] assert [k for k in b['data'] if k not in moved] == [k for k in c['data'] if k not in moved] member_order_variances.append(label) if label == 'R5_STATUS_EXISTING_GOAL': assert b['data']['id'] == finals['B']['rows']['user_weight_status'][0]['id'] assert c['data']['id'] == finals['C']['rows']['user_weight_status'][0]['id'] c['data']['id'] = b['data']['id'] assert semantic_equal(b, c), label pairs += 1 assert pairs == 8 and set(member_order_variances) == {'W1_MANUAL', 'W2_GOAL'} for left, right in [(1, 1.0), (0, False), (None, ''), ([], {}), ([1, 2], [2, 1])]: assert not semantic_equal(left, right) return {'full_semantic_pairs': pairs, 'object_member_order_only_variances': member_order_variances, 'preserved': 'all field names, native JSON types, scalar values, nulls, array order; raw object order retained', 'mapped': 'only source-proven own user_id/nick_name/goal identity and validated trace_id'} def verify(): manifest = tool.ROOT / BASE / 'manifest.json' assert hashlib.sha256(manifest.read_bytes()).hexdigest() == 'd313205f4377b037d2bd6cdfee4200b28ebed1373234eb0a7c44c38796429a55' root = tool.ROOT / BASE / 'execution' result = json.loads(subprocess.run([sys.executable, '-B', str(root / 'audit.py')], check=True, capture_output=True, text=True).stdout) assert result['checks'] == 2615 for lane, sql in [('B', 425), ('C', 589)]: facts = result['lanes'][lane] assert facts['HTTP'] == facts['DTO_full_pass'] == 8 assert facts['prepare_INSERT'] == 2 and facts['formal_INSERT'] == 7 and facts['formal_UPDATE'] == 3 assert facts['marked_rows'] == 9 and facts['total_SQL'] == sql assert facts['old_and_unrelated_protection'] and facts['resources_cleaned'] seals = json.loads(subprocess.run([sys.executable, '-B', str(root / 'verify_seal.py')], check=True, capture_output=True, text=True).stdout) assert seals['all_seals_immutable'] and seals['verified_assets'] == 903 assert seals['manifest_SHA'] == hashlib.sha256(manifest.read_bytes()).hexdigest() cross_audit() source = json.loads((root / 'source-frozen-sha.json').read_text())['files'] assert len(source) == 6168 for relative, digest in source.items(): assert hashlib.sha256((tool.ROOT.parent / 'zhctproject/ai_api' / relative).read_bytes()).hexdigest() == digest def old_method_targets(): text = (tool.ROOT / OLD).read_text() result = [] for action in ['info', 'weightManageInfo']: assert '`POST /api/user.userWeight/' + action + '`' in text result.append(item('ai_api', 'POST', '/api/user.UserWeight/' + action, BASE, evidence=OLD, evidence_only=True, risk='historical-client-POST-method-preserved-no-HTTP-replay')) return result def targets(): result = [] for method, action in [('POST', 'setWeight'), ('POST', 'weightManage'), ('GET', 'byDate'), ('GET', 'info'), ('GET', 'weightManageInfo'), ('GET', 'status'), ('GET', 'dates')]: extras = {} if action in ['info', 'weightManageInfo']: extras.update(register_new_runtime_key=True, runtime_observed_method=method, runtime_inventory_scope='controller-direct-http-observed', runtime_direct_http_proof=BASE + 'REPORT.md', runtime_controller_source='zhctproject/ai_api/app/api/controller/user/UserWeight.php') result.append(item('ai_api', method, '/api/user.UserWeight/' + action, BASE, scope='customer-AI-own-weight-goal-score-bind-health-DB-chain-Store-sync-explicit-failure-only', risk='owned-nonempty-real-weight-goal-sidecar-write-readback', request_type='application/json-form' if method == 'POST' else 'GET-query', parity='partial-evidence-eight-full-JSON-semantic-pairs-native-types-and-array-order-equal-object-member-order-variance-documented', success_case='partial-evidence-AI-local-owned-nonempty-chain-real-Store-sync-pending', boundary_empty_min_max='partial-evidence-nonempty-manual-and-goal-overwrite-same-day-only', duplicate_retry='not-tested', cleanup='nine-owned-marked-business-rows-retained-one-token-two-cache-workers-ports-released', **extras)) return result if __name__ == '__main__': methods = Path(__file__).with_name('integration-20261011-batch85-methods.json') audit = Path(__file__).with_name('integration-20261011-batch85.json') assert not methods.exists() and not audit.exists(), 'Preserve one-time audit transition' verify() tool.main(old_method_targets(), methods, lambda: None) tool.main(targets(), audit, lambda: None)