#!/usr/bin/env python3 """Merchant seven-column repair: preserve failed index-name assertion, verify followup.""" import hashlib import json import re from pathlib import Path import integrate_20261010 as tool from integrate_20261011_batches46_48 import item HERE = Path(__file__).resolve().parent BASE = 'work/2026-09-28-smart-canteen-dm8-poc/store-seven-columns-20261012/' PRIVATE = Path('/Users/wiscourper/Documents/workspace/local/dm8-batch116-20261012.S1TtvG/store-seven-columns') HTTP = Path('/private/tmp/cdx-ai-hk-course-b.Kmdirn/runtime') def pinned(root, name, digest): raw = (root / name).read_bytes() assert hashlib.sha256(raw).hexdigest() == digest, name return json.loads(raw) def verify(): before = pinned(PRIVATE, 'inspect.json', 'c5152a09911bbb0271cd3f535659dce35e40232dc6a2b76e7bf7be5f3d521a50') shadow = pinned(PRIVATE, 'shadow.json', '0d477ba2f01fad07a400a1e39448f740faeb3a0855559cb258ce12b400d0bd7b') apply = pinned(PRIVATE, 'apply.json', '066b01ecd0180f23860bb2a278f2adda7437def34bd003bd8c4fe854a97a6603') post = pinned(PRIVATE, 'postcheck.json', '20599b18d199dcf743208be4e5d670be1c8094b7e0868a281b63f55ed0a44aa9') assert before['status'] == shadow['status'] == 'passed' and shadow['assertions'] == 40 assert before['before']['row_count'] == 5 and len(before['before']['columns']) == 11 assert apply['status'] == 'failed' and apply['error'] == 'indexes unchanged' and apply['assertions'] == 27 assert post['passed'] and len(post['checks']) == 8 and all(post['checks'].values()) assert post['original_rows_sha256'] == before['before']['rows_sha256'] assert post['defaults_row_count'] == 5 and len(post['columns']) == 18 logs = {} for name, digest, count in [ ('shadow.jsonl', 'e51cab904f3954e3f3cc037dd407944d8f95cf40063822875698b39d47e362e6', 40), ('apply.jsonl', '87d7f578040bbfdb22a22082a146f7f0bc1a26d84653bccbee6b72994b08ca06', 26), ]: raw = (PRIVATE / name).read_bytes() assert hashlib.sha256(raw).hexdigest() == digest logs[name] = [json.loads(line) for line in raw.splitlines()] assert sum(e.get('passed') is True for e in logs[name]) == count executed = [e for e in logs['apply.jsonl'] if e.get('phase') == 'before-sql'] recorded_sql = '\n'.join(line for line in (tool.ROOT / BASE / 'applied.sql').read_text().splitlines() if not line.startswith('--')).strip().rstrip(';') normalize_sql = lambda sql: re.sub(r'\s*([(),])\s*', r'\1', ' '.join(sql.split())) assert len(executed) == 1 and normalize_sql(executed[0]['sql']) == normalize_sql(recorded_sql) assert [e['ok'] for e in logs['apply.jsonl'] if e.get('phase') == 'after-sql'] == [True] assert any(e.get('check') == 'original column definitions preserved' and e['passed'] for e in logs['apply.jsonl']) lengths = {'GROUP_NAME': 64, 'TYPES': 128, 'CONTACT_NAME': 32, 'CONTACT_PHONE': 20, 'ADDRESS': 255} for col in post['columns'][11:]: name = col['column_name'] assert col['nullable'] == 'N' if name in lengths: assert col['data_type'] == 'VARCHAR' and col['char_length'] == lengths[name] assert col['char_used'] == 'C' and col['data_default'] == "''" else: assert name in ('GROUP_ID', 'STATUS') assert col['data_type'] == ('BIGINT' if name == 'GROUP_ID' else 'SMALLINT') assert col['data_default'] == ('0' if name == 'GROUP_ID' else '1') assert {'index_name': 'INDEX33555722', 'uniqueness': 'UNIQUE'} in before['before']['indexes'] assert {'index_name': 'INDEX33555722', 'uniqueness': 'UNIQUE', 'index_type': 'NORMAL', 'status': 'VALID'} in post['indexes'] assert post['index_columns'] == [{'index_name': 'INDEX33555722', 'column_name': 'STORE_ID', 'column_position': 1}] assert post['constraints'] == [{'constraint_name': 'CONS134218923', 'constraint_type': 'P', 'status': 'ENABLED'}] assert post['constraint_columns'] == [{'constraint_name': 'CONS134218923', 'column_name': 'STORE_ID', 'position': 1}] assert len(post['indexes']) == 2 and sum(i['index_type'] == 'CLUSTER' and i['status'] == 'VALID' for i in post['indexes']) == 1 samples = [] for name, digest in [ ('store-info-seven-retest-CDXMODE104DBF24A63.json', '5e8ee74ccf15a69fcbca0057415cf903c86fedc8b317f7d1f1c48316c7d6fa39'), ('C_store_info_seven_retest_CD5B763DAE977B.json', 'afa33558e7aea67c2ec484ad86d17d76873bb433bd756eddecf62bdf30665ae2'), ]: events = pinned(HTTP, name, digest)['events'] rows = [e for e in events if 'path' in e] assert {e['path'] for e in rows} == {'/store/Store/info', '/terminal/Store/info'} assert len(rows) == 2 and events[-1]['token_absent'] for e in rows: assert e['method'] == 'GET' and e['http'] == e['business_status'] == 200 and e['pass'] assert e['source_row_count'] == 1 and e['source_rows_unchanged'] and e['external_stub_delta_bytes'] == 0 assert e['store_info_field_count'] == 19 types = e['seven_field_presence_and_json_type'] assert set(types) == {'group_id', 'group_name', 'types', 'status', 'contact_name', 'contact_phone', 'address'} for k, v in types.items(): assert v == {'present': True, 'json_type': 'integer' if k in ('group_id', 'status') else 'string'} assert e['core11_row_fingerprint'] == events[0]['source_row_fingerprints']['core11']['row_fingerprints'][0] samples.append({e['path']: e for e in rows}) for path in samples[0]: assert samples[0][path]['response_shape'] == samples[1][path]['response_shape'] if __name__ == '__main__': audit = HERE / 'integration-20261012-batch127.json' assert not audit.exists(), 'Preserve prior transition audit; import verify for read-only checking' targets = [item('ai_api', 'GET', '/' + app + '/Store/info', BASE, from_repo='unreviewed', from_source='apifox-2026-05-08', scope='bounded-existing-merchant-read-after-approved-DM-seven-column-repair', success_case='partial-evidence-nonempty-merchant19fields-seven-new-field-types-aligned', parity='partial-evidence-full-native-type-tree-equal-different-source-snapshots-not-value-equal', readback='tested-pass-own-source-row-fingerprint-unchanged-and-DDL-original-eleven-fields-preserved', cleanup='read-only-http-existing-business-values-preserved-zero-external-temporary-token-removed') for app in ['store', 'terminal']] tool.main(targets, audit, verify)