#!/usr/bin/env python3 from __future__ import annotations import csv import hashlib import json import py_compile import re import shutil import sys import xml.etree.ElementTree as ET from pathlib import Path ROOT = Path(__file__).resolve().parents[1] ERRORS: list[str] = [] WARNINGS: list[str] = [] CHECKS: list[str] = [] def ok(message: str) -> None: CHECKS.append(message) def fail(message: str) -> None: ERRORS.append(message) def warn(message: str) -> None: WARNINGS.append(message) def active_files(pattern: str): for path in ROOT.rglob(pattern): rel = path.relative_to(ROOT) if rel.parts[:1] == ('attachments',): continue if rel.parts[:2] == ('research', 'original'): continue if any(part in {'bin', 'obj', 'node_modules', '__pycache__', '.git'} for part in rel.parts): continue yield path required = [ 'README.md', 'AGENTS.md', 'global.json', 'KArtSell.sln', 'RELEASE_NOTES_V12_4.md', 'docs/v12_4/00_EXECUTIVE_CONTINUOUS_MODEL_OPERATIONS.md', 'docs/v12_4/01_SOURCE_COVERAGE_MATRIX.csv', 'docs/v12_4/07_SLICE_CATALOGUE.csv', 'docs/v12_4/09_DETAILED_WBS_MASTER.csv', 'docs/v12_4/10_TECH_DEBT_REGISTER.csv', 'docs/v12_4/11_TRACEABILITY_MATRIX.csv', 'docs/v12_4/12_DECISION_LOG.csv', 'docs/v12_4/16_VALIDATION_MATRIX.csv', 'docs/v12_4/K-ArtSell_Aegis_v12_4_지속모델운영_통합고도화_제안서.docx', 'db/migrations/0016_continuous_model_operations.sql', 'contracts/policies/sell-policy-contract.v1.json', 'contracts/events/signal-decision-created.v2.schema.json', 'contracts/metrics/outcome-metrics.v2.json', 'contracts/schedules/model-operations.v1.json', 'contracts/model-governance/promotion-gate.v1.json', 'src/KArtSell.Modules.ModelOperations/KArtSell.Modules.ModelOperations.csproj', 'tests/KArtSell.ModelOperations.UnitTests/KArtSell.ModelOperations.UnitTests.csproj', 'scripts/scaffold_slice.py', 'scripts/tests/test_scaffold_slice.py', 'templates/dotnet/VerticalSlice/Endpoint.cs.template', 'templates/vue/Feature/schema.ts.template', ] for name in required: if not (ROOT/name).exists(): fail(f'missing required file: {name}') if not ERRORS: ok(f'required files present: {len(required)}') # Source preservation and exact SHA coverage. coverage_path = ROOT/'docs/v12_4/01_SOURCE_COVERAGE_MATRIX.csv' coverage: dict[str, str] = {} if coverage_path.exists(): with coverage_path.open(encoding='utf-8-sig', newline='') as f: for row in csv.DictReader(f): coverage[row['Attachment']] = row['SHA256'] attachment_dir = ROOT/'attachments/current_session' attachment_files = [p for p in sorted(attachment_dir.glob('*')) if p.is_file()] for path in attachment_files: actual = hashlib.sha256(path.read_bytes()).hexdigest() if coverage.get(path.name) != actual: fail(f'attachment coverage/hash mismatch: {path.name}') if len(coverage) != len(attachment_files): fail('source coverage row count differs from current-session attachment count') if len(attachment_files) != 8: fail(f'expected exactly eight current-session attachments, found {len(attachment_files)}') else: ok('eight-file cumulative source preservation verified') # Syntax and tabular structure. for path in active_files('*.json'): try: json.loads(path.read_text(encoding='utf-8')) except Exception as exc: fail(f'invalid JSON {path.relative_to(ROOT)}: {exc}') ok('JSON parse complete') for path in [*active_files('*.csproj'), *active_files('*.props')]: try: ET.parse(path) except Exception as exc: fail(f'invalid XML {path.relative_to(ROOT)}: {exc}') ok('MSBuild XML parse complete') for path in active_files('*.csv'): try: with path.open(encoding='utf-8-sig', newline='') as f: rows=list(csv.reader(f)) if not rows: fail(f'empty CSV {path.relative_to(ROOT)}'); continue width=len(rows[0]); bad=[i+1 for i,row in enumerate(rows) if len(row)!=width] if bad: fail(f'ragged CSV {path.relative_to(ROOT)} rows {bad[:10]}') except Exception as exc: fail(f'invalid CSV {path.relative_to(ROOT)}: {exc}') ok('CSV structure complete') wbs_path=ROOT/'docs/v12_4/09_DETAILED_WBS_MASTER.csv' if wbs_path.exists(): with wbs_path.open(encoding='utf-8-sig', newline='') as f: wbs=list(csv.DictReader(f)) ids=[r['WBS_ID'] for r in wbs] if len(ids)!=len(set(ids)): fail('duplicate WBS_ID') if len(wbs)!=288: fail(f'v12.4 WBS must contain 288 tasks, found {len(wbs)}') slice_ids={m.group(0) for r in wbs for m in re.finditer(r'VS-\d{2}', r['Slice_ID'])} expected={f'VS-{i:02d}' for i in range(26)} if slice_ids!=expected: fail(f'WBS slice coverage mismatch: missing={sorted(expected-slice_ids)} extra={sorted(slice_ids-expected)}') required_columns={'Requirement_ID','API_ID','DB_Migration_ID','Job_Event_ID','UI_ID','Test_ID','Acceptance_Evidence','Primary_Owner','Dependency','Gate','Evidence_Class'} if not required_columns.issubset(wbs[0]): fail('WBS required columns missing') if any(not r['Acceptance_Evidence'].strip() or not r['Primary_Owner'].strip() for r in wbs): fail('WBS has blank acceptance or owner') ok('WBS uniqueness/coverage complete: 288 tasks, 26 slices') for path in active_files('*.py'): try: py_compile.compile(str(path), doraise=True) except Exception as exc: fail(f'Python compile failed {path.relative_to(ROOT)}: {exc}') ok('Python compile complete') for path in active_files('*.yaml'): text=path.read_text(encoding='utf-8') if '\t' in text: fail(f'YAML contains tab indentation: {path.relative_to(ROOT)}') if not text.strip(): fail(f'empty YAML: {path.relative_to(ROOT)}') ok('YAML basic checks complete') # Source anti-patterns and module boundary checks. source_files=[p for p in [*active_files('*.cs'),*active_files('*.sql')] if 'tests' not in p.relative_to(ROOT).parts] patterns={'AllowAnonymous()':'anonymous module endpoint','IGenericRepository':'generic repository','DateTime.Now':'direct wall-clock use','DateTime.UtcNow':'direct wall-clock use','dynamic ':'dynamic typing in production C#'} for token,label in patterns.items(): hits=[str(p.relative_to(ROOT)) for p in source_files if token in p.read_text(encoding='utf-8')] if hits: fail(f'{label} pattern {token}: {hits}') select_star=[str(p.relative_to(ROOT)) for p in source_files if re.search(r'\bselect\s+\*',p.read_text(encoding='utf-8'),flags=re.I)] if select_star: fail(f'SELECT * prohibited: {select_star}') ok('prohibited source pattern scan complete') for path in active_files('Endpoint.cs'): text=path.read_text(encoding='utf-8') if 'Roles(' not in text and 'Policies(' not in text: fail(f'endpoint lacks Roles/Policies: {path.relative_to(ROOT)}') ok('endpoint authorization declarations complete') for path in active_files('*.cs'): rel=path.relative_to(ROOT) if 'Domain' in rel.parts: text=path.read_text(encoding='utf-8') for token in ['using Dapper','using Npgsql','using FastEndpoints','using Hangfire','HttpContext','DbConnection']: if token in text: fail(f'domain framework dependency {token}: {rel}') ok('domain dependency scan complete') for path in active_files('*.cs'): text=path.read_text(encoding='utf-8') scrub=re.sub(r'""".*?"""','',text,flags=re.S) scrub=re.sub(r'"(?:\\.|[^"\\])*"','',scrub) if scrub.count('{') != scrub.count('}'): fail(f'C# brace imbalance: {path.relative_to(ROOT)}') ok('C# structural smoke complete') migration_names=sorted(p.name for p in (ROOT/'db/migrations').glob('*.sql')) if not all(re.match(r'^\d{4}_[a-z0-9_]+\.sql$',n) for n in migration_names): fail('migration filename format must be NNNN_snake_case.sql') if len(migration_names)!=len(set(migration_names)): fail('duplicate migration filename') nums=[int(n[:4]) for n in migration_names] if nums != sorted(nums): fail('migration sequence is not ordered') if '0016_continuous_model_operations.sql' not in migration_names: fail('continuous model operations migration missing') ok(f'migration ordering complete: {len(migration_names)} scripts') # Fail-closed capabilities. program=(ROOT/'src/KArtSell.Host/Program.cs').read_text(encoding='utf-8') for marker in ['!x.AutomaticOrder','!x.KisOrderAdapter','ValidateOnStart']: if marker not in program: fail(f'missing startup capability guard: {marker}') for cfg in [ROOT/'src/KArtSell.Host/appsettings.json',ROOT/'src/KArtSell.Host/appsettings.Development.json']: if cfg.exists(): data=json.loads(cfg.read_text(encoding='utf-8')); caps=data.get('Capabilities') if caps is not None and (caps.get('AutomaticOrder') is not False or caps.get('KisOrderAdapter') is not False): fail(f'order capability is not OFF: {cfg.relative_to(ROOT)}') ok('automatic-order and KIS capability guards complete') # Policy registry <-> C# contract cross-check. policy=json.loads((ROOT/'contracts/policies/sell-policy-contract.v1.json').read_text(encoding='utf-8')) contract=(ROOT/'src/KArtSell.Modules.SignalEngine/Domain/SellPolicyContract.cs').read_text(encoding='utf-8') def const_string(name: str) -> str | None: m=re.search(rf'public const string {re.escape(name)} = "([^"]+)";',contract); return m.group(1) if m else None def const_int(name: str) -> int | None: m=re.search(rf'public const int {re.escape(name)} = (\d+);',contract); return int(m.group(1)) if m else None def const_decimal(name: str) -> float | None: m=re.search(rf'public const decimal {re.escape(name)} = ([0-9.]+)m;',contract); return float(m.group(1)) if m else None if const_string('ContractVersion') != policy['contractVersion']: fail('C# and JSON policy contract version drift') if const_string('DecisionContractVersion') != policy['decisionContractVersion']: fail('decision contract version drift') if const_int('PolicyTraceSchemaVersion') != policy['policyTraceSchemaVersion']: fail('policy trace schema version drift') name_map={ 'ALG-SELL-001':('HardImpairmentPolicyId','HardImpairmentPriority'), 'ALG-SELL-PORT-001':('PortfolioSurvivalPolicyId','PortfolioSurvivalPriority'), 'ALG-SELL-002':('GapFloorBreachPolicyId','GapFloorBreachPriority'), 'ALG-SELL-003':('TwoCloseFloorBreachPolicyId','TwoCloseFloorBreachPriority'), 'ALG-SELL-004':('ConcentrationLiquidityPolicyId','ConcentrationLiquidityPriority'), 'ALG-SELL-005':('OpportunityCostPolicyId','OpportunityCostPriority')} for item in policy['policies']: id_name, pri_name=name_map[item['policyId']] if const_string(id_name)!=item['policyId'] or const_int(pri_name)!=item['priority']: fail(f'policy registry drift: {item["policyId"]}') ratio_checks=[('HardImpairmentSellRatioOfLot',1.0),('GapFloorAtrThreshold',1.5),('GapFloorSellRatioOfLot',0.4),('TwoCloseSellRatioOfLot',0.2),('OpportunityMinimumSellRatioOfLot',0.1),('OpportunityMaximumSellRatioOfLot',0.25)] for name,expected in ratio_checks: if const_decimal(name)!=expected: fail(f'policy decimal drift: {name}') ok('machine-readable policy registry matches C# contract') # v12.4 semantic safety, API/FE contract and defect regression tokens. migration=(ROOT/'db/migrations/0015_signal_engine_semantic_versioning.sql').read_text(encoding='utf-8') reader=(ROOT/'src/KArtSell.Modules.SignalEngine/Infrastructure/DapperSellDecisionContextReader.cs').read_text(encoding='utf-8') service=(ROOT/'src/KArtSell.Modules.SignalEngine/Application/SellDecisionService.cs').read_text(encoding='utf-8') opp=(ROOT/'src/KArtSell.Modules.SignalEngine/Domain/Policies/OpportunityCostPolicy.cs').read_text(encoding='utf-8') unit=(ROOT/'tests/KArtSell.SignalEngine.UnitTests/SellPolicyChainTests.cs').read_text(encoding='utf-8') fe=(ROOT/'frontend/src/features/sell-decision/schema.ts').read_text(encoding='utf-8') for token in ['weight_semantics_version','decision_contract_version','policy_trace_schema_version','policy_contract_definition']: if token not in migration: fail(f'missing 0015 migration contract token: {token}') if 'weight_semantics_version = 2' not in reader: fail('legacy ambiguous decision context is not blocked') for token in ['DecisionContractVersion','PolicyTraceSchemaVersion']: if token not in service: fail(f'decision evidence contract token missing: {token}') if 'OPPORTUNITY_RATIO_NOT_POSITIVE' not in opp or 'Positive_opportunity_edge_with_zero_requested_ratio' not in unit: fail('zero-ratio opportunity sell regression is not closed') for token in ["decisionContractVersion: z.literal('sell-decision.v2')",'policyTraceSchemaVersion: z.literal(2)']: if token not in fe: fail(f'frontend runtime contract token missing: {token}') if not (ROOT/'frontend/src/features/sell-decision/components/PolicyTracePanel.vue').exists(): fail('PolicyTracePanel component missing') ok('v12.4 semantic-version, contract-version and zero-ratio safety checks complete') # CI and scaffolding controls. ci=(ROOT/'.gitea/workflows/ci.yml').read_text(encoding='utf-8') if 'python scripts/validate_v124.py' not in ci: fail('CI does not execute v12.4 validator') if re.search(r'python scripts/validate_v12[123]\.py', ci): fail('CI references stale validator') # Detect duplicate consecutive working-directory keys in a step. if re.search(r'working-directory:[^\n]+\n\s*working-directory:',ci): fail('duplicate working-directory key in CI') scaffold=(ROOT/'scripts/scaffold_slice.py').read_text(encoding='utf-8') for token in ['refusing to overwrite existing files','SCAFFOLD_ONLY','DRY_RUN']: if token not in scaffold: fail(f'scaffolder fail-closed token missing: {token}') ok('CI current-validator and deterministic scaffold controls complete') # v12.4 continuous model operations contract and safety checks. schedule_contract=json.loads((ROOT/'contracts/schedules/model-operations.v1.json').read_text(encoding='utf-8')) if schedule_contract.get('contractVersion') != 'model-operations.v1': fail('model operations contract version mismatch') allowed={'EVALUATION_ONLY','PROPOSAL_ONLY','DRILL_ONLY'} operations=schedule_contract.get('operations',[]) codes=[op.get('operationCode') for op in operations] if len(codes)!=10 or len(codes)!=len(set(codes)): fail('model operation code count/uniqueness mismatch') if set(codes)!={'J10','J11','J17','J18','J19','J20','J21','J22','J23','J24'}: fail('model operation registry code set mismatch') if any(op.get('mode') not in allowed for op in operations): fail('forbidden scheduler automation mode present') for forbidden in ['AUTO_PROMOTE','AUTO_ROLLBACK','AUTO_PARAMETER_CHANGE','AUTO_ORDER','KIS_SUBMISSION']: if forbidden not in schedule_contract['automationBoundary']['forbidden']: fail(f'missing forbidden automation boundary: {forbidden}') registry=(ROOT/'src/KArtSell.Modules.ModelOperations/Domain/ModelOperationDefinition.cs').read_text(encoding='utf-8') for op in operations: for token in [op['operationCode'],op['name'],op['queue']]: if token not in registry: fail(f'C#/JSON model operation registry drift: {token}') migration=(ROOT/'db/migrations/0016_continuous_model_operations.sql').read_text(encoding='utf-8') for token in ['model_operation_schedule','model_operation_request','model_operation_status_event','model_metric_definition','model_metric_observation','model_evaluation_snapshot','model_improvement_proposal','model_promotion_review','model_rollback_drill']: if token not in migration: fail(f'0016 missing model operations object: {token}') if 'AUTO_PROMOTE' in migration or 'AUTO_ORDER' in migration: fail('migration contains forbidden automatic mutation mode') service=(ROOT/'src/KArtSell.Modules.ModelOperations/Application/ModelOperationRequestService.cs').read_text(encoding='utf-8') repo=(ROOT/'src/KArtSell.Modules.ModelOperations/Infrastructure/DapperModelOperationRequestRepository.cs').read_text(encoding='utf-8') for token in ['VersionSet','ToContractValue','ReadAsync']: if token not in service: fail(f'model operation request service missing frozen-context token: {token}') for token in ['ModelOperationRequested','IOutboxWriter','idempotency_key','model_operation_status_event']: if token not in repo: fail(f'model operation request/outbox token missing: {token}') gate=(ROOT/'src/KArtSell.Modules.ModelOperations/Domain/PromotionGateEvaluator.cs').read_text(encoding='utf-8') for token in ['RequiresIndependentValidation: true','RequiresHumanApproval: true','EVIDENCE_ONLY_NO_AUTO_PROMOTION']: if token not in gate: fail(f'promotion gate safety token missing: {token}') frontend_schema=(ROOT/'frontend/src/features/model-operations/schema.ts').read_text(encoding='utf-8') for token in ["'EVALUATION_ONLY'", "'PROPOSAL_ONLY'", "'DRILL_ONLY'", "EVALUATION_AND_PROPOSAL_ONLY_HUMAN_APPROVAL_REQUIRED"]: if token not in frontend_schema: fail(f'frontend model operations boundary token missing: {token}') if 'AUTO_PROMOTE' in frontend_schema: fail('frontend schema permits automatic promotion') with (ROOT/'docs/v12_4/10_TECH_DEBT_REGISTER.csv').open(encoding='utf-8-sig', newline='') as f: debts=list(csv.DictReader(f)) if len(debts)!=52: fail(f'v12.4 technical debt register must contain 52 items, found {len(debts)}') with (ROOT/'docs/v12_4/12_DECISION_LOG.csv').open(encoding='utf-8-sig', newline='') as f: decisions=list(csv.DictReader(f)) if len(decisions)!=32: fail(f'v12.4 decision log must contain 32 items, found {len(decisions)}') ok('v12.4 scheduler, metric, frozen-context and no-auto-mutation checks complete') if not (ROOT/'frontend/pnpm-lock.yaml').exists(): warn('frontend/pnpm-lock.yaml missing; G0 must generate, review and commit it') if shutil.which('dotnet') is None: warn('dotnet SDK unavailable; restore/build/test not executed') if shutil.which('pnpm') is None: warn('pnpm unavailable; frozen install/typecheck/test/build not executed') if shutil.which('psql') is None: warn('PostgreSQL client unavailable; migration rehearsal not executed') print('K-ArtSell Aegis v12.4 static validation') print(f'root={ROOT}') for m in CHECKS: print(f'PASS: {m}') for m in WARNINGS: print(f'WARN: {m}') for m in ERRORS: print(f'FAIL: {m}') print(f'summary: pass={len(CHECKS)} warn={len(WARNINGS)} fail={len(ERRORS)}') sys.exit(1 if ERRORS else 0)