#!/usr/bin/env python3 """Single-writer, evidence-checked update. Never converts partial coverage to closed.""" import csv import hashlib import io import json from pathlib import Path import subprocess import sys HERE = Path(__file__).resolve().parent ROOT = HERE.parents[2] COVERAGE = ROOT / 'work/2026-09-28-smart-canteen-dm8-poc/http-cron-coverage-matrix.csv' CONTRACT = ROOT / 'work_store/apifox/2026-09-30-dm8-interface-acceptance/derived/interface-contract-matrix.csv' AI = 'work_ai_api/2026-09-30-dm8-interface-acceptance/' STORE = 'work_store/apifox/2026-09-30-dm8-interface-acceptance/' def source_root(repo): """Resolve active isolated code, never the user's freely switched checkout.""" profile = json.loads((HERE.parent / 'ACTIVE_WORKTREES.json').read_text()) spec = profile['repositories'][repo] relative = Path(spec['workspace_relative']) if relative.is_absolute() or '..' in relative.parts or relative.parts[0] != 'worktrees': raise RuntimeError('Unsafe isolated worktree path') target = ROOT.parent / relative if target.is_symlink() or not (target / '.git').is_file(): raise RuntimeError('Expected an independent Git worktree: ' + str(target)) branch = subprocess.check_output(['git', '-C', str(target), 'branch', '--show-current'], text=True).strip() if branch != spec['branch']: raise RuntimeError('Isolated test branch changed: ' + repo) subprocess.run(['git', '-C', str(target), 'merge-base', '--is-ancestor', spec['base_commit'], 'HEAD'], check=True) return target def load(relative): return json.loads((ROOT / relative).read_text()) def verify_evidence(): for lane in ('B', 'C'): r = load(STORE + f'parallel-store-consume-limit-20261010/result-{lane}.json') assert r['status'] == 'pass' and len(r['samples']) == 17 and all(r['checks'].values()) r = load(AI + f'parallel-ai-category-crud-20261010/evidence-green-{lane}.json') assert r['case_count'] == 18 and len(r['form_encoding_scenarios']) == 4 and r['error'] is None assert r['cleanup'] == {'fresh_pdo_owned_rows': 0, 'owned_relation_rows': 0, 'token_absent': True, 'existing_categories_unchanged': True} r = load(AI + 'parallel-ai-terminal-catalog-20261010/green_results.json') assert r['unique_get_routes'] == 12 and r['scenario_pairs'] == 15 assert r['mysql_source_unchanged'] and r['dm_source_unchanged'] and r['tokens_cleaned'] assert all(s['type_differences'] == 0 and s['mysql_source_projection_equal'] and s['dm_source_projection_equal'] for s in r['scenarios']) for lane in ('B', 'C'): r = load(AI + f'parallel-ai-goods-service-crud-20261010/evidence-{lane}.json') assert r['case_count'] == 30 and r['error'] is None assert r['cleanup'] == {'fresh_pdo_owned_rows': 0, 'owned_relation_rows': 0, 'token_absent': True, 'natural_tables_unchanged': True} r = load(AI + 'parallel-ai-help-crud-20261010/results.json') assert r['scenario_pairs'] == r['positive_source_cases'] == r['full_response_strict_equal'] == 84 assert r['owned_record_differences'] == 0 and r['exact_method_path_keys'] == 8 for audit in r['audit'].values(): assert audit['natural_unchanged'] and audit['independent_remaining'] == 0 and audit['token_cleaned'] for relative in [AI + 'parallel-ai-gift-wide-20261010/compare.py', STORE + 'bill-counter-diagnosis-20261010/verify_results.py', STORE + 'parallel-store-finance-reads-20261010/compare.py']: subprocess.run([sys.executable, str(ROOT / relative)], check=True, capture_output=True) def norm(path): return '/' + path.split('?', 1)[0].strip('/').lower() def targets(): out = [] def add(repo, method, path, module, terminal, role, risk, scope, evidence, **extra): out.append(dict(repo=repo, method=method, path=path, module=module, terminal=terminal, role=role, risk=risk, scope=scope, evidence=evidence, **extra)) for action in ('lists', 'create', 'update', 'detail', 'changeStatus', 'delete'): extra = {} if action in ('create', 'delete'): extra['missing_required'] = 'partial-evidence-required-fields-rejected' if action in ('update', 'changeStatus'): extra['invalid_type'] = 'partial-evidence-invalid-account-or-status-no-write' if action in ('update', 'delete'): extra['duplicate_retry'] = 'partial-evidence-repeat-edit-or-delete-readback' if action == 'lists': extra['boundary_empty_min_max'] = 'partial-evidence-page2-empty-and-filter-miss' add('store', 'POST', '/api/consumeLimit/' + action, '消费限制管理', 'PC端', 'isolated-disabled-admin', 'money-policy-config', 'store-primary-owned-limit-five-tables', STORE + 'parallel-store-consume-limit-20261010/REPORT.md', request_type='application/x-www-form-urlencoded', **extra) for app, terminal in [('store', 'PC端'), ('terminal', '设备端')]: for action in ('list', 'add', 'edit', 'delete'): add('ai_api', 'GET' if action == 'list' else 'POST', f'/{app}/Category/{action}', '商品分类', terminal, 'isolated-merchant-cache-token', 'catalog-write' if action != 'list' else 'read', 'ai-primary-own-category-and-goods-reference-read', AI + 'parallel-ai-category-crud-20261010/REPORT.md', request_type='JSON-and-form-see-sampled-edit-evidence' if action == 'edit' else ('GET-query' if action == 'list' else 'application/json'), boundary_empty_min_max='partial-evidence-own-parent-child-tree-and-cycle-or-parent-delete-rejection') rows = load(AI + 'parallel-ai-terminal-catalog-20261010/green_results.json')['scenarios'] for path in sorted({r['path'] for r in rows}): historical = path.endswith('/dishes.Foods/list') or path.endswith('/dishes.Foods/detail') add('ai_api', 'GET', path, '设备食物菜品字典', '设备端', 'isolated-merchant-cache-token', 'read', 'ai-primary-readonly-catalog-ten-source-tables', AI + 'parallel-ai-terminal-catalog-20261010/REPORT.md', parity='partial-evidence-types-equal-source-verified-historical-value-variance' if historical else 'partial-evidence-full-sample-response-equal', boundary_empty_min_max='not-tested') for action in ('detail', 'list'): add('ai_api', 'GET', '/api/gift.Log/' + action, '赠送金账单', '移动端', 'synthetic-mobile-user-own-token', 'money-read', 'ai-primary-own-ledger-store-auth-read', AI + 'parallel-ai-gift-wide-20261010/REPORT.md', boundary_empty_min_max='partial-evidence-zero-small-2pow53-plus1-phpmax-uint64max', parity='partial-evidence-six-full-responses-equal-with-stable-tie-order') for action in ('list', 'detail'): add('store', 'GET', '/api/cashRefund/' + action, '手机退款读取', '移动端', 'existing-nonproduction-user-readonly-token', 'money-read', 'customer-store-ai-refund-auth-readonly', STORE + 'parallel-store-finance-reads-20261010/REPORT.md', parity='partial-evidence-source-values-verified-and-JSON-types-equal', boundary_empty_min_max='partial-evidence-missing-refund-error' if action == 'detail' else 'not-tested') add('store', 'GET', '/api/finance/visitor/flows', '访客流水读取', '外部接口', 'existing-nonproduction-api-auth-readonly', 'money-read', 'customer-store-ai-visitor-ledger-readonly', STORE + 'parallel-store-finance-reads-20261010/REPORT.md', mysql_state='partial-evidence-private-debug-disabled-narrow-filter', performance='not-accepted-wide-query-and-original-debug-mode-open', mysql_readback='partial-evidence-private-debug-disabled-source-subset', parity='partial-evidence-source-subset-and-JSON-types-equal-cursor-no-duplicates', boundary_empty_min_max='partial-evidence-invalid-date-and-zero-page-size') for app, terminal in [('store', 'PC端'), ('terminal', '设备端')]: for action in ('list', 'all', 'add', 'edit', 'delete'): add('ai_api', 'GET' if action in ('list', 'all') else 'POST', f'/{app}/goods.Service/{action}', '商品服务字典', terminal, 'isolated-merchant-cache-token', 'catalog-write' if action in ('add', 'edit', 'delete') else 'read', 'ai-primary-own-goods-service-and-reference-read', AI + 'parallel-ai-goods-service-crud-20261010/REPORT.md', boundary_empty_min_max='partial-evidence-empty-summary-quoted-Chinese-pagination-sort', duplicate_retry='partial-evidence-duplicate-add-and-repeat-soft-delete', request_type='GET-query' if action in ('list', 'all') else 'JSON-or-form-see-evidence') for action in ('list', 'add', 'edit', 'delete'): add('ai_api', 'GET' if action == 'list' else 'POST', f'/{app}/content.Help/{action}', '帮助文档', terminal, 'isolated-merchant-cache-token', 'content-write' if action != 'list' else 'read', 'ai-primary-own-help-document', AI + 'parallel-ai-help-crud-20261010/REPORT.md', boundary_empty_min_max='partial-evidence-empty-quoted-Chinese-32767-32768-32769-49152-byte-text', request_type='GET-query' if action == 'list' else 'application/json') add('store', 'POST', '/p/bill/add', '菜谱保存', 'PC端', 'isolated-disabled-admin', 'write', 'store-primary-own-recipes-dishes-version', STORE + 'bill-counter-diagnosis-20261010/REPORT.md', duplicate_retry='partial-evidence-repeat-save-and-clear-no-count-drift', request_type='application/x-www-form-urlencoded-with-data-JSON-string', boundary_empty_min_max='partial-evidence-zero-cache-existing-recipe-and-clear-preserve-other-stall', parity='partial-evidence-seven-full-responses-equal-no-transaction-change') return out def read_csv(path): raw = path.read_bytes() reader = csv.DictReader(io.StringIO(raw.decode('utf-8'), newline='')) return list(reader.fieldnames), list(reader), raw def append_evidence(old, new): parts = old.split(';') if old else [] if new not in parts: parts.append(new) return ';'.join(parts) def main(specs=None, audit_path=None, verifier=None): (verifier or verify_evidence)() specs = targets() if specs is None else specs for target in specs: if not (ROOT / target['evidence']).is_file(): raise RuntimeError('Evidence report not ready: ' + target['evidence']) audit_path = HERE / 'integration-20261010.json' if audit_path is None else Path(audit_path) previous = json.loads(audit_path.read_text()) if audit_path.exists() else {} resolution = previous.get('method_resolution', []) ownership = previous.get('ownership_resolution', []) aliases = previous.get('case_alias_resolution', []) registrations = previous.get('runtime_key_registration', []) result = {'scope': 'sampled business paths only; all coverage gates remain open', 'method_resolution': resolution, 'ownership_resolution': ownership, 'case_alias_resolution': aliases, 'runtime_key_registration': registrations, 'files': {}, 'target_count': len(specs)} pending = [] for path, column in [(COVERAGE, 'route_or_command'), (CONTRACT, 'path')]: headers, rows, raw = read_csv(path) for target in specs: if not target.get('register_new_runtime_key'): continue direct_report = Path(target.get('runtime_direct_http_proof', '')) direct_observed = ( target.get('runtime_inventory_scope') == 'controller-direct-http-observed' and bool(target.get('runtime_direct_http_proof')) and not direct_report.is_absolute() and '..' not in direct_report.parts and (ROOT / direct_report).is_file() ) if (target['method'] not in ['GET','POST','PUT','PATCH','DELETE','HEAD','OPTIONS'] or target.get('runtime_observed_method') != target['method'] or (not target.get('runtime_client_source') and not direct_observed) or not target.get('runtime_controller_source')): raise RuntimeError('New runtime key requires actual method and client/controller proof') if any(row['repo'] in (target['repo'],'unreviewed') and norm(row[column]) == norm(target['path']) and row['method'] in (target['method'],'*') for row in rows): raise RuntimeError('New runtime key already registered or ambiguous') added = dict.fromkeys(headers,'') added.update(repo=target['repo'],method=target['method'], source='controller-direct-http-runtime-observed' if direct_observed else 'current-client-controller-runtime-observed') added[column] = target['path'] if column == 'route_or_command': added.update(kind='http',mysql_http_or_command='not-tested',dm_http_or_command='not-tested',database_readback='not-tested') else: added['coverage_gate']='open' rows.append(added) registrations.append({'file':str(path.relative_to(ROOT)),'initial_registered_row':dict(added), 'actual_method':target['method'],'client_source':target.get('runtime_client_source',''), 'direct_http_proof':str(direct_report) if direct_observed else '', 'controller_source':target['runtime_controller_source'],'evidence':target['evidence'], 'meaning':('Previously absent from this inventory; actual direct HTTP and current Controller observed. Product-client usage is unverified; not a static-public-method census.' if direct_observed else 'Previously absent from this inventory; current client, controller and real HTTP evidence required. Not a static-public-method census.')}) verified_aliases = set() for target in specs: declared = target.get('verified_source_case_aliases') if not declared: continue original = [dict(row) for row in rows if row['repo'] in (target['repo'], target.get('from_repo')) and row['method'] == target['method'] and norm(row[column]) == norm(target['path'])] # One-time ownership resolution only. No row deletion, generalized # underscore aliasing, or absorption of an already-owned endpoint. if (len(declared) < 2 or len({name.lower() for name in declared}) != 1 or len(original) != len(declared) or {row[column] for row in original} != set(declared) or any(row['repo'] != target.get('from_repo') or row['source'] != target.get('from_source') for row in original)): raise RuntimeError('Case alias evidence does not cover original rows') verified_aliases.add((target['repo'], target['method'], norm(target['path']))) aliases.append({'file': str(path.relative_to(ROOT)), 'original_rows': original, 'verified_source_case_aliases': declared, 'evidence': target['evidence'], 'meaning': 'Case-only raw labels retained; normalized method/path key is counted once, only the reported actual URL was requested.'}) matched = set() for row in rows: if column == 'route_or_command' and row['kind'] != 'http': continue for target in specs: if row['repo'] not in (target['repo'], target.get('from_repo')) or norm(row[column]) != norm(target['path']): continue if row['method'] not in (target['method'], '*'): continue if row['repo'] != target['repo']: if row['source'] != target['from_source']: raise RuntimeError('Unexpected original owner/source') if any(r is not row and r['repo'] == target['repo'] and r['method'] == target['method'] and norm(r[column]) == norm(target['path']) for r in rows) and ( target['repo'], target['method'], norm(target['path'])) not in verified_aliases: raise RuntimeError('Ownership assignment would create a duplicate key') ownership.append({'file': str(path.relative_to(ROOT)), 'old_row': dict(row), 'observed_repo': target['repo'], 'observed_method': target['method'], 'evidence': target['evidence'], 'meaning': 'Actual source and private HTTP dispatch verified; same path in another repo remains independent.'}) row['repo'] = target['repo'] row['source'] = 'runtime-source-owner-verified-from-apifox' if row['method'] == '*': if row['source'] != 'controller-action-source': raise RuntimeError('Do not resolve a genuine ANY route as one observed method') item = {'file': str(path.relative_to(ROOT)), 'repo': row['repo'], 'path': row[column], 'old_method': '*', 'observed_method': target['method'], 'old_row': dict(row), 'meaning': 'Previously unspecified source candidate; only this HTTP method observed. Other verbs not claimed.'} resolution.append(item) row['method'] = target['method'] matched.add((target['repo'], target['method'], norm(target['path']))) row['module'], row['risk'] = target['module'], target['risk'] row['evidence'] = append_evidence(row.get('evidence', ''), target['evidence']) if target.get('evidence_only'): continue # Preserve stronger/different prior scenario status. if column == 'route_or_command': row.update(db_access_review=target['scope'], mysql_http_or_command=target.get('mysql_state', 'tested-pass'), dm_http_or_command=target.get('dm_state', 'tested-pass'), database_readback=target.get('readback', 'tested-pass-scoped-own-row-or-source-projection-and-independent-audit'), external_stub='private-worker-real-stubs-no-external-effects') else: delivery_scope = target.get('delivery_scope', row['delivery_scope'] if row.get('delivery_scope', '').startswith('confirmed-') else 'database-adaptation-source-route-observed') row.update(delivery_scope=delivery_scope, terminal=target['terminal'], role=target['role'], reachability='observed-' + target['method'] + '-B-C-only-other-methods-not-claimed', db_scope=target['scope'], request_types=target.get('request_type', 'GET-query' if target['method'] == 'GET' else 'application/json'), success_case=target.get('success_case', 'partial-evidence-B-C-sampled-business-paths'), mysql_input_and_readback=target.get('mysql_readback', 'tested-pass-code-local-scoped-original-table-readback'), dm_input_and_readback=target.get('dm_readback', 'tested-pass-nonproduction-scoped-original-table-readback'), response_parity=target.get('parity', 'partial-evidence-scoped-response-types-and-values-equal'), fixture='owned-marker-or-readonly-source-see-report', cleanup=target.get('cleanup', 'independent-zero-or-source-snapshot-unchanged'), coverage_gate='open') for field in ('missing_required', 'invalid_type', 'boundary_empty_min_max', 'duplicate_retry', 'performance'): if field in target: row[field] = target[field] expected = {(t['repo'], t['method'], norm(t['path'])) for t in specs} if matched != expected: raise RuntimeError('Missing registered target: ' + repr(expected - matched)) stream = io.StringIO(newline='') writer = csv.DictWriter(stream, fieldnames=headers, lineterminator='\r\n' if b'\r\n' in raw else '\n') writer.writeheader(); writer.writerows(rows) new = stream.getvalue().encode('utf-8') pending.append((path, raw, new)) result['files'][str(path.relative_to(ROOT))] = {'before_sha256': hashlib.sha256(raw).hexdigest(), 'after_sha256': hashlib.sha256(new).hexdigest(), 'rows': len(rows), 'matched_targets': len(matched)} # Validate both inputs before changing either shared matrix. if any(path.read_bytes() != raw for path, raw, _ in pending): raise RuntimeError('Concurrent matrix writer detected') for path, _, new in pending: path.write_bytes(new) audit_path.write_text(json.dumps(result, ensure_ascii=False, indent=2) + '\n') print(json.dumps({'updated_targets': len(specs), 'resolved_unspecified_methods': len(resolution), 'resolved_ownership_rows': len(ownership), 'closed_gates_added': 0, 'audit': str(audit_path.relative_to(ROOT))})) if __name__ == '__main__': main()