Initial commit: Add project files
ci / backend (push) Failing after 12s
ci / frontend (push) Failing after 19s
ci / static (push) Failing after 45s

This commit is contained in:
2026-08-02 05:15:36 +09:00
commit dcd1322d41
636 changed files with 122352 additions and 0 deletions
View File
+33
View File
@@ -0,0 +1,33 @@
#!/usr/bin/env python3
from __future__ import annotations
import hashlib, json
from pathlib import Path
ROOT=Path(__file__).resolve().parents[1]
EXCLUDE={'PACKAGE_MANIFEST.json','SHA256SUMS.txt'}
def sha(path: Path)->str:
h=hashlib.sha256()
with path.open('rb') as f:
for b in iter(lambda:f.read(1024*1024),b''): h.update(b)
return h.hexdigest()
files=[]
for p in sorted(ROOT.rglob('*')):
if not p.is_file(): continue
rel=p.relative_to(ROOT).as_posix()
if rel in EXCLUDE or '/__pycache__/' in f'/{rel}/': continue
files.append({'path':rel,'bytes':p.stat().st_size,'sha256':sha(p)})
manifest={
'package':'KArtSell_Aegis_v12_5_execution_assurance_complete',
'version':'12.5',
'status':'IMPLEMENTATION_TEMPLATE_STATIC_VALIDATED_RUNTIME_EVIDENCE_REQUIRED',
'boundary':['RESEARCH_CANDIDATE_NOT_PRODUCTION','AUTOMATIC_ORDER_OFF','KIS_SUBMISSION_OFF','AUTO_MODEL_PROMOTION_OFF'],
'fileCount':len(files),
'totalBytes':sum(x['bytes'] for x in files),
'files':files,
}
(ROOT/'PACKAGE_MANIFEST.json').write_text(json.dumps(manifest,ensure_ascii=False,indent=2)+'\n',encoding='utf-8')
with (ROOT/'SHA256SUMS.txt').open('w',encoding='utf-8') as f:
for x in files: f.write(f"{x['sha256']} {x['path']}\n")
print(f"manifest files={len(files)} bytes={manifest['totalBytes']}")
+108
View File
@@ -0,0 +1,108 @@
#!/usr/bin/env python3
from __future__ import annotations
import argparse
import json
import re
import sys
from dataclasses import dataclass
from pathlib import Path
ROOT = Path(__file__).resolve().parents[1]
PASCAL = re.compile(r"^[A-Z][A-Za-z0-9]*$")
REQ = re.compile(r"^REQ-[A-Z0-9-]+$")
ROUTE = re.compile(r"^[a-z0-9][a-z0-9_/{}/.-]*$")
ALLOWED_METHODS = {"Get", "Post", "Put", "Patch", "Delete"}
ALLOWED_ROLES = {"Admin", "Advisor", "Approver", "Auditor", "Compliance", "DataOps", "Quant", "Risk", "System"}
@dataclass(frozen=True)
class Spec:
module: str
slice_name: str
requirement_id: str
http_method: str
route: str
role: str
frontend_feature: str
def validate(spec: Spec) -> None:
errors: list[str] = []
if not PASCAL.fullmatch(spec.module): errors.append("module must be PascalCase")
if not PASCAL.fullmatch(spec.slice_name): errors.append("slice must be PascalCase")
if not REQ.fullmatch(spec.requirement_id): errors.append("requirement must match REQ-[A-Z0-9-]+")
if spec.http_method not in ALLOWED_METHODS: errors.append(f"method must be one of {sorted(ALLOWED_METHODS)}")
if not ROUTE.fullmatch(spec.route) or spec.route.startswith("/"): errors.append("route must be relative lower-case API path")
if spec.role not in ALLOWED_ROLES: errors.append(f"role must be one of {sorted(ALLOWED_ROLES)}")
if not re.fullmatch(r"[a-z][a-z0-9-]*", spec.frontend_feature): errors.append("feature must be kebab-case")
if errors: raise ValueError("; ".join(errors))
def render(template: Path, values: dict[str, str]) -> str:
text = template.read_text(encoding="utf-8")
for key, value in values.items(): text = text.replace(f"__{key}__", value)
unresolved = sorted(set(re.findall(r"__[A-Z0-9_]+__", text)))
if unresolved: raise ValueError(f"unresolved template tokens in {template.name}: {unresolved}")
return text
def plan(spec: Spec) -> dict[Path, str]:
validate(spec)
values = {
"MODULE": f"KArtSell.Modules.{spec.module}",
"SLICE": spec.slice_name,
"REQUIREMENT_ID": spec.requirement_id,
"HTTP_METHOD": spec.http_method,
"ROUTE": spec.route,
"ROLE": spec.role,
"FEATURE": spec.frontend_feature,
}
backend_dir = ROOT / "src" / f"KArtSell.Modules.{spec.module}" / "Features" / spec.slice_name
frontend_dir = ROOT / "frontend" / "src" / "features" / spec.frontend_feature
files: dict[Path, str] = {}
dotnet = ROOT / "templates" / "dotnet" / "VerticalSlice"
for name in ["Endpoint.cs", "Request.cs", "Response.cs", "Handler.cs", "Validator.cs", "Sql.cs", "README.md"]:
files[backend_dir / name] = render(dotnet / f"{name}.template", values)
vue = ROOT / "templates" / "vue" / "Feature"
for template_name, output_name in [
("api.ts.template", "api.ts"), ("queries.ts.template", "queries.ts"),
("schema.ts.template", "schema.ts"), ("page.vue.template", "pages/FeaturePage.vue"),
("schema.spec.ts.template", "tests/schema.spec.ts")]:
files[frontend_dir / output_name] = render(vue / template_name, values)
return files
def main() -> int:
parser = argparse.ArgumentParser(description="Create a fail-closed Vertical Slice scaffold from reviewed templates.")
parser.add_argument("--module", required=True)
parser.add_argument("--slice", dest="slice_name", required=True)
parser.add_argument("--requirement", required=True)
parser.add_argument("--method", default="Post")
parser.add_argument("--route", required=True)
parser.add_argument("--role", required=True)
parser.add_argument("--feature", required=True)
parser.add_argument("--write", action="store_true", help="write files; otherwise print a deterministic plan")
args = parser.parse_args()
try:
spec = Spec(args.module, args.slice_name, args.requirement, args.method, args.route, args.role, args.feature)
files = plan(spec)
collisions = [str(path.relative_to(ROOT)) for path in files if path.exists()]
if collisions: raise FileExistsError("refusing to overwrite existing files: " + ", ".join(collisions))
manifest = {str(path.relative_to(ROOT)): content for path, content in sorted(files.items())}
if not args.write:
print(json.dumps({"status": "DRY_RUN", "files": list(manifest)}, indent=2, ensure_ascii=False))
return 0
for path, content in files.items():
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text(content, encoding="utf-8")
marker = ROOT / "docs" / "generated" / f"{spec.requirement_id}_{spec.slice_name}.json"
marker.parent.mkdir(parents=True, exist_ok=True)
marker.write_text(json.dumps({"status":"SCAFFOLD_ONLY","spec":spec.__dict__,"files":list(manifest)}, ensure_ascii=False, indent=2)+"\n", encoding="utf-8")
print(json.dumps({"status": "WRITTEN", "files": list(manifest), "marker": str(marker.relative_to(ROOT))}, indent=2, ensure_ascii=False))
return 0
except (ValueError, FileExistsError) as exc:
print(f"ERROR: {exc}", file=sys.stderr)
return 2
if __name__ == "__main__":
raise SystemExit(main())
View File
+25
View File
@@ -0,0 +1,25 @@
from pathlib import Path
import sys
import unittest
ROOT = Path(__file__).resolve().parents[2]
sys.path.insert(0, str(ROOT / "scripts"))
from scaffold_slice import Spec, plan, validate # noqa: E402
class ScaffoldSliceTests(unittest.TestCase):
def test_plan_is_deterministic_and_contains_be_fe_contracts(self):
spec = Spec("MarketData", "IngestDailyBars", "REQ-DAT-001", "Post", "internal/v1/data/bars", "DataOps", "market-data")
first = plan(spec)
second = plan(spec)
self.assertEqual(first, second)
paths = {str(path.relative_to(ROOT)) for path in first}
self.assertIn("src/KArtSell.Modules.MarketData/Features/IngestDailyBars/Endpoint.cs", paths)
self.assertIn("frontend/src/features/market-data/schema.ts", paths)
self.assertTrue(all("__" not in content for content in first.values()))
def test_invalid_or_unsafe_spec_is_rejected(self):
with self.assertRaises(ValueError):
validate(Spec("marketData", "Slice", "REQ-X", "Post", "/unsafe", "Unknown", "Bad_Feature"))
if __name__ == "__main__":
unittest.main()
+17
View File
@@ -0,0 +1,17 @@
from pathlib import Path
import subprocess
import tempfile
import unittest
class ScaffoldUiScreenTests(unittest.TestCase):
def test_tool_contains_vendor_boundary(self):
root = Path(__file__).resolve().parents[2]
text = (root / 'tools/scaffold_ui_screen.py').read_text(encoding='utf-8')
self.assertIn('standard UI screen without vendor imports', text)
for template in (root / 'templates/vue/screens').glob('*.template'):
body = template.read_text(encoding='utf-8')
self.assertNotIn("from 'primevue", body)
self.assertNotIn("from 'ag-grid", body)
if __name__ == '__main__':
unittest.main()
+174
View File
@@ -0,0 +1,174 @@
#!/usr/bin/env python3
from __future__ import annotations
import csv
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 rel.parts[:1] == ("tests",):
continue
if any(part in {"bin", "obj", "node_modules", "__pycache__"} for part in rel.parts):
continue
yield path
required = [
"README.md",
"AGENTS.md",
"global.json",
"KArtSell.sln",
"docs/v12/00_EXECUTIVE_INTEGRATED_PROPOSAL.md",
"docs/v12/07_WBS_MASTER.csv",
"docs/v12/08_TECH_DEBT_REGISTER.csv",
"docs/v12/09_TRACEABILITY_MATRIX.csv",
"db/migrations/0012_signal_engine_integrated_hardening.sql",
"research/hardening/TEST_RESULTS.txt",
"attachments/original/K-ArtSell_12_2_complete_package(1).zip",
"attachments/original/K-ArtSell_퀀트투자자문_SI_본프로그램착수_누적고도화_통합실행기준서_v10.0(1).docx",
"attachments/original/KArtSell_v11_implementation_acceleration(1).zip",
"attachments/original/KArtSell_Aegis_v11_1_hardening(1).zip",
]
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)}")
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 handle:
rows = list(csv.reader(handle))
if not rows:
fail(f"empty CSV {path.relative_to(ROOT)}")
else:
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")
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_files = [*active_files("*.cs"), *active_files("*.sql")]
patterns = {
"AllowAnonymous()": "anonymous module endpoint",
"IGenericRepository": "generic repository",
"DateTime.Now": "direct wall-clock use",
"DateTime.UtcNow": "direct wall-clock use",
}
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 = []
for path in source_files:
text = path.read_text(encoding="utf-8")
if re.search(r"\bselect\s+\*", text, flags=re.IGNORECASE):
select_star.append(str(path.relative_to(ROOT)))
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"):
if f"{Path('Domain')}" in str(path.relative_to(ROOT)):
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}: {path.relative_to(ROOT)}")
ok("domain dependency scan 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$", name) for name 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")
ok(f"migration ordering complete: {len(migration_names)} scripts")
for path in active_files("*"):
if path.is_file() and path.name.lower() in {"testfile", "temp", "tmp"}:
fail(f"placeholder file: {path.relative_to(ROOT)}")
ok("placeholder scan complete")
if not (ROOT / "frontend" / "pnpm-lock.yaml").exists():
warn("frontend/pnpm-lock.yaml is missing by design; G0 must generate and review it")
if shutil.which("dotnet") is None:
warn("dotnet SDK unavailable in current validation environment")
if shutil.which("pnpm") is None:
warn("pnpm unavailable in current validation environment")
print("K-ArtSell Aegis v12.0 static validation")
print(f"root={ROOT}")
for message in CHECKS:
print(f"PASS: {message}")
for message in WARNINGS:
print(f"WARN: {message}")
for message in ERRORS:
print(f"FAIL: {message}")
print(f"summary: pass={len(CHECKS)} warn={len(WARNINGS)} fail={len(ERRORS)}")
sys.exit(1 if ERRORS else 0)
+189
View File
@@ -0,0 +1,189 @@
#!/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',
'docs/v12_1/00_EXECUTIVE_EXECUTION_PROPOSAL.md',
'docs/v12_1/01_SOURCE_COVERAGE_MATRIX.csv',
'docs/v12_1/06_SLICE_CATALOGUE.csv',
'docs/v12_1/07_DETAILED_WBS_MASTER.csv',
'docs/v12_1/08_TECH_DEBT_REGISTER.csv',
'docs/v12_1/09_TRACEABILITY_MATRIX.csv',
'docs/v12_1/10_DECISION_LOG.csv',
'db/migrations/0013_execution_readiness_building_blocks.sql',
'research/hardening/test_policy_contract.py',
]
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)}')
# All current-session attachments must be byte-for-byte represented by source coverage.
coverage_path = ROOT/'docs/v12_1/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'
for path in sorted(attachment_dir.glob('*')):
if not path.is_file(): continue
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([p for p in attachment_dir.glob('*') if p.is_file()]):
fail('source coverage row count differs from current-session attachment count')
else:
ok(f'attachment preservation verified: {len(coverage)} files')
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 structural and traceability checks.
wbs_path=ROOT/'docs/v12_1/07_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)<180: fail(f'detailed WBS is unexpectedly small: {len(wbs)}')
slice_ids={r['Slice_ID'] for r in wbs if r['Slice_ID'].startswith('VS-')}
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(f'WBS uniqueness/coverage complete: {len(wbs)} 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_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=[]
for path in source_files:
if re.search(r'\bselect\s+\*',path.read_text(encoding='utf-8'),flags=re.I): select_star.append(str(path.relative_to(ROOT)))
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')
# Lightweight brace balance catches truncated generated source without pretending to compile C#.
for path in active_files('*.cs'):
text=path.read_text(encoding='utf-8')
# Remove raw/normal string bodies sufficiently for a structural smoke check.
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')
ok(f'migration ordering complete: {len(migration_names)} scripts')
for path in active_files('*'):
if path.is_file() and path.name.lower() in {'testfile','temp','tmp'}: fail(f'placeholder file: {path.relative_to(ROOT)}')
ok('placeholder scan complete')
# Safety capability must be fail-closed in code and configuration.
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')
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.1 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)
+216
View File
@@ -0,0 +1,216 @@
#!/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',
'docs/v12_2/00_EXECUTIVE_STRATEGIC_PROPOSAL.md',
'docs/v12_2/01_SOURCE_COVERAGE_MATRIX.csv',
'docs/v12_2/06_SLICE_CATALOGUE.csv',
'docs/v12_2/07_DETAILED_WBS_MASTER.csv',
'docs/v12_2/08_TECH_DEBT_REGISTER.csv',
'docs/v12_2/09_TRACEABILITY_MATRIX.csv',
'docs/v12_2/10_DECISION_LOG.csv',
'db/migrations/0014_signal_engine_lot_weight_and_policy_trace.sql',
'research/hardening/test_policy_contract.py',
'templates/dotnet/VerticalSlice/Endpoint.cs.template',
'templates/vue/Feature/schema.ts.template',
'templates/sql/Migration.sql.template',
'templates/hangfire/ReliableJob.cs.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)}')
# All current-session attachments must be byte-for-byte represented by source coverage.
coverage_path = ROOT/'docs/v12_2/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'
for path in sorted(attachment_dir.glob('*')):
if not path.is_file(): continue
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([p for p in attachment_dir.glob('*') if p.is_file()]):
fail('source coverage row count differs from current-session attachment count')
else:
ok(f'attachment preservation verified: {len(coverage)} files')
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 structural and traceability checks.
wbs_path=ROOT/'docs/v12_2/07_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)<225: fail(f'detailed WBS is unexpectedly small: {len(wbs)}')
slice_ids={r['Slice_ID'] for r in wbs if r['Slice_ID'].startswith('VS-')}
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(f'WBS uniqueness/coverage complete: {len(wbs)} 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_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=[]
for path in source_files:
if re.search(r'\bselect\s+\*',path.read_text(encoding='utf-8'),flags=re.I): select_star.append(str(path.relative_to(ROOT)))
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')
# Lightweight brace balance catches truncated generated source without pretending to compile C#.
for path in active_files('*.cs'):
text=path.read_text(encoding='utf-8')
# Remove raw/normal string bodies sufficiently for a structural smoke check.
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')
ok(f'migration ordering complete: {len(migration_names)} scripts')
for path in active_files('*'):
if path.is_file() and path.name.lower() in {'testfile','temp','tmp'}: fail(f'placeholder file: {path.relative_to(ROOT)}')
ok('placeholder scan complete')
# Safety capability must be fail-closed in code and configuration.
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')
# v12.2 data-semantics and traceability checks.
input_text=(ROOT/'src/KArtSell.Modules.SignalEngine/Domain/SellDecisionInput.cs').read_text(encoding='utf-8')
service_text=(ROOT/'src/KArtSell.Modules.SignalEngine/Application/SellDecisionService.cs').read_text(encoding='utf-8')
fe_schema=(ROOT/'frontend/src/features/sell-decision/schema.ts').read_text(encoding='utf-8')
migration=(ROOT/'db/migrations/0014_signal_engine_lot_weight_and_policy_trace.sql').read_text(encoding='utf-8')
for token in ['CurrentSecurityPortfolioWeight','CurrentLotPortfolioWeight','TargetSecurityPortfolioWeightAfter']:
if token not in input_text and token not in service_text:
fail(f'missing v12.2 explicit weight token: {token}')
for token in ['policy_trace_json','SignalDecisionCreated",\n 2']:
if token not in service_text:
fail(f'missing v12.2 policy trace/event token: {token}')
for token in ['currentSecurityPortfolioWeight','currentLotPortfolioWeight','targetSecurityPortfolioWeightAfter','policyTrace']:
if token not in fe_schema:
fail(f'missing frontend runtime contract token: {token}')
for token in ['current_security_portfolio_weight','current_lot_portfolio_weight','policy_trace_json']:
if token not in migration:
fail(f'missing migration token: {token}')
if len(coverage) != 6:
fail(f'expected exactly six current-session attachments, found {len(coverage)}')
else:
ok('v12.2 explicit lot/security weight and policy-trace contracts 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.2 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)
+240
View File
@@ -0,0 +1,240 @@
#!/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_3.md',
'docs/v12_3/00_EXECUTIVE_DELIVERY_HARDENING.md',
'docs/v12_3/01_SOURCE_COVERAGE_MATRIX.csv',
'docs/v12_3/06_SLICE_CATALOGUE.csv',
'docs/v12_3/07_DETAILED_WBS_MASTER.csv',
'docs/v12_3/08_TECH_DEBT_REGISTER.csv',
'docs/v12_3/09_TRACEABILITY_MATRIX.csv',
'docs/v12_3/10_DECISION_LOG.csv',
'docs/v12_3/15_VALIDATION_MATRIX.csv',
'docs/v12_3/K-ArtSell_Aegis_v12_3_실행검증_통합고도화_제안서.docx',
'db/migrations/0015_signal_engine_semantic_versioning.sql',
'contracts/policies/sell-policy-contract.v1.json',
'contracts/events/signal-decision-created.v2.schema.json',
'contracts/metrics/outcome-metrics.v1.json',
'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_3/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) != 7:
fail(f'expected exactly seven current-session attachments, found {len(attachment_files)}')
else:
ok('seven-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_3/07_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)!=250: fail(f'v12.3 WBS must contain 250 tasks, found {len(wbs)}')
slice_ids={r['Slice_ID'] for r in wbs if r['Slice_ID'].startswith('VS-')}
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: 250 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 '0015_signal_engine_semantic_versioning.sql' not in migration_names: fail('semantic version 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.3 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.3 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_v123.py' not in ci: fail('CI does not execute v12.3 validator')
if 'validate_v121.py' in ci or 'validate_v122.py' in 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')
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.3 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)
+285
View File
@@ -0,0 +1,285 @@
#!/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)
+202
View File
@@ -0,0 +1,202 @@
#!/usr/bin/env python3
from __future__ import annotations
import csv
import hashlib
import json
import py_compile
import re
import sys
import xml.etree.ElementTree as ET
from pathlib import Path
ROOT = Path(__file__).resolve().parents[1]
ERRORS: list[str] = []
WARNINGS: list[str] = []
PASSES: list[str] = []
def ok(message: str) -> None: PASSES.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', 'RELEASE_NOTES_V12_5.md', 'SOURCE_BASIS_V12_5.md',
'docs/v12_5/00_EXECUTIVE_CRITICAL_REVIEW.md',
'docs/v12_5/01_SOURCE_COVERAGE_MATRIX.csv',
'docs/v12_5/02_ROLE_BASED_BRUTAL_AUDIT.md',
'docs/v12_5/03_TARGET_ARCHITECTURE_SOLID_REFACTORING.md',
'docs/v12_5/04_ALGORITHM_MODEL_EVALUATION.md',
'docs/v12_5/05_DATA_DB_INTEGRITY.md',
'docs/v12_5/06_BE_FE_HANGFIRE_IMPLEMENTATION.md',
'docs/v12_5/07_QA_SECURITY_OPERATIONS.md',
'docs/v12_5/08_PROCESS_SIMPLIFICATION_VIBE_CODING.md',
'docs/v12_5/09_ROADMAP_20D_90D_36M.md',
'docs/v12_5/10_DETAILED_WBS_MASTER.csv',
'docs/v12_5/11_TECH_DEBT_REGISTER.csv',
'docs/v12_5/12_TRACEABILITY_MATRIX.csv',
'docs/v12_5/13_DECISION_LOG.csv',
'docs/v12_5/14_PACKAGE_CONTENTS.md',
'docs/v12_5/15_JOB_CATALOGUE.csv',
'docs/v12_5/16_RACI_MATRIX.csv',
'docs/v12_5/17_RISK_REGISTER.csv',
'docs/v12_5/18_VALIDATION_MATRIX.csv',
'docs/v12_5/K-ArtSell_Aegis_v12_5_실행보증_통합고도화_제안서.docx',
'db/migrations/0017_execution_assurance.sql',
'contracts/schedules/execution-assurance.v1.json',
'contracts/model-governance/evidence-classification.v1.json',
'src/KArtSell.Modules.ModelOperations/Domain/ModelOperationExecutionBoundary.cs',
'tests/KArtSell.ModelOperations.UnitTests/ModelOperationExecutionBoundaryTests.cs',
]
for name in required:
if not (ROOT / name).exists(): fail(f'missing required file: {name}')
if not ERRORS: ok(f'required v12.5 files present: {len(required)}')
# Current request direct attachment preservation.
coverage_path = ROOT / 'docs/v12_5/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_request_20260802'
attachments = [p for p in sorted(attachment_dir.iterdir()) if p.is_file()]
if len(attachments) != 3:
fail(f'expected exactly three direct current-request attachments, found {len(attachments)}')
for path in attachments:
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(attachments): fail('source coverage row count mismatch')
if not ERRORS: ok('three direct attachments preserved byte-for-byte')
# Parse structured files.
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 MSBuild 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')
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')
# WBS/debt/trace dimensions.
def read_dicts(rel: str):
with (ROOT / rel).open(encoding='utf-8-sig', newline='') as f: return list(csv.DictReader(f))
wbs = read_dicts('docs/v12_5/10_DETAILED_WBS_MASTER.csv') if (ROOT/'docs/v12_5/10_DETAILED_WBS_MASTER.csv').exists() else []
if len(wbs) != 336: fail(f'v12.5 WBS must contain 336 rows, found {len(wbs)}')
ids = [r.get('WBS_ID','') for r in wbs]
if len(ids) != len(set(ids)): fail('duplicate WBS_ID')
if sum(1 for r in wbs if r.get('WBS_ID','').startswith('AEG-V12-5-')) != 48: fail('v12.5 WBS delta must contain 48 rows')
if any(not r.get('Acceptance_Evidence','').strip() or not r.get('Primary_Owner','').strip() for r in wbs): fail('WBS blank acceptance/owner')
ok('WBS 336 rows, unique IDs, 48-row delta')
debt = read_dicts('docs/v12_5/11_TECH_DEBT_REGISTER.csv') if (ROOT/'docs/v12_5/11_TECH_DEBT_REGISTER.csv').exists() else []
if len(debt) != 68: fail(f'tech debt must contain 68 rows, found {len(debt)}')
if len({r.get('ID') for r in debt}) != len(debt): fail('duplicate tech debt ID')
ok('technical debt register 68 rows')
trace = read_dicts('docs/v12_5/12_TRACEABILITY_MATRIX.csv') if (ROOT/'docs/v12_5/12_TRACEABILITY_MATRIX.csv').exists() else []
for i in range(1, 13):
if not any(r.get('Requirement_ID') == f'REQ-EXA-{i:03d}' for r in trace): fail(f'missing trace requirement REQ-EXA-{i:03d}')
ok('execution-assurance trace requirements complete')
# Fail-closed configuration and code markers.
settings = json.loads((ROOT/'src/KArtSell.Host/appsettings.json').read_text(encoding='utf-8'))
if settings.get('ModelOperations', {}).get('DispatcherEnabled') is not False: fail('ModelOperations dispatcher must default OFF')
program = (ROOT/'src/KArtSell.Host/Program.cs').read_text(encoding='utf-8')
for marker in ['modelOperationsDispatcherEnabled', '!modelOperationsDispatcherEnabled || x.ShadowEvaluation',
'RegisterModelOperationsSchedules(modelOperationsDispatcherEnabled', '!x.AutomaticOrder', '!x.KisOrderAdapter']:
if marker not in program: fail(f'missing fail-closed marker: {marker}')
scheduler = (ROOT/'src/KArtSell.Modules.ModelOperations/Scheduling/ModelOperationsScheduler.cs').read_text(encoding='utf-8')
for marker in ['if (!dispatcherEnabled)', 'RemoveIfExists', 'DispatcherJobId']:
if marker not in scheduler: fail(f'missing scheduler safety marker: {marker}')
contract = (ROOT/'contracts/schedules/execution-assurance.v1.json')
if contract.exists():
c = json.loads(contract.read_text(encoding='utf-8'))
if c.get('dispatcher', {}).get('enabledByDefault') is not False: fail('schedule contract must default dispatcher OFF')
if len(c.get('operations', [])) != 6: fail('execution-assurance contract must contain J25-J30')
ok('fail-closed model-operations boundary complete')
registry = (ROOT/'src/KArtSell.Modules.ModelOperations/Domain/ModelOperationDefinition.cs').read_text(encoding='utf-8')
codes = re.findall(r'new\("(J\d+)"', registry)
if len(codes) != 16 or len(set(codes)) != 16: fail(f'model operation registry expected 16 unique operations, found {len(codes)}/{len(set(codes))}')
for code in [f'J{i}' for i in range(25,31)]:
if code not in codes: fail(f'missing registry operation {code}')
ok('model operation registry J10-J30 approved set complete')
migration = (ROOT/'db/migrations/0017_execution_assurance.sql').read_text(encoding='utf-8')
for marker in ['V12_5_REAPPROVAL_REQUIRED', 'model_operation_artifact', 'release_evidence_bundle',
'source_contract_snapshot', 'execution_assurance_decision', "'J30'"]:
if marker not in migration: fail(f'migration 0017 missing marker: {marker}')
ok('migration 0017 execution-assurance structures complete')
# Prohibited patterns in active source.
source_files = [p for p in [*active_files('*.cs'), *active_files('*.sql')] if 'tests' not in p.relative_to(ROOT).parts]
patterns = {'AllowAnonymous()':'anonymous endpoint','IGenericRepository':'generic repository','DateTime.Now':'direct wall clock','DateTime.UtcNow':'direct wall clock'}
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}: {token} in {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-pattern scan complete')
# Domain framework dependency scan and C# structural smoke.
for path in active_files('*.cs'):
rel = path.relative_to(ROOT)
text = path.read_text(encoding='utf-8')
if 'Domain' in rel.parts:
for token in ['using Dapper', 'using Npgsql', 'using FastEndpoints', 'using Hangfire', 'HttpContext', 'DbConnection']:
if token in text: fail(f'domain framework dependency {token}: {rel}')
scrub = re.sub(r'""".*?"""', '', text, flags=re.S)
scrub = re.sub(r'"(?:\\.|[^"\\])*"', '', scrub)
if scrub.count('{') != scrub.count('}'): fail(f'C# brace imbalance: {rel}')
ok('domain dependency and C# structural smoke complete')
# Package status honesty.
claims = []
for path in [ROOT/'README.md', ROOT/'RELEASE_NOTES_V12_5.md', ROOT/'docs/v12_5/00_EXECUTIVE_CRITICAL_REVIEW.md']:
text = path.read_text(encoding='utf-8')
for forbidden in ['PRODUCTION_READY', 'BUILD_VALIDATED / DB_REHEARSED', 'SHADOW_VALIDATED']:
if forbidden in text: claims.append(f'{path.name}:{forbidden}')
if claims: fail(f'unsupported readiness claim: {claims}')
else: ok('readiness claims remain evidence-bounded')
# Toolchain warnings are expected in this container/package.
if not (ROOT/'frontend/pnpm-lock.yaml').exists(): warn('pnpm-lock.yaml is still absent; frozen frontend validation remains blocked')
warn('.NET 10 restore/build/test is not executed by static validator')
warn('PostgreSQL DbUp fresh/upgrade/re-run/failure rehearsal is not executed by static validator')
warn('252-trading-day Shadow, PBO/DSR, security and DR evidence remain unmet')
print(f'PASS={len(PASSES)} WARN={len(WARNINGS)} FAIL={len(ERRORS)}')
for m in PASSES: print(f'PASS: {m}')
for m in WARNINGS: print(f'WARN: {m}')
for m in ERRORS: print(f'FAIL: {m}')
sys.exit(1 if ERRORS else 0)
+15
View File
@@ -0,0 +1,15 @@
#!/usr/bin/env bash
set -euo pipefail
command -v dotnet >/dev/null || { echo "dotnet SDK 10 is required" >&2; exit 1; }
command -v node >/dev/null || { echo "Node.js is required" >&2; exit 1; }
command -v corepack >/dev/null || { echo "corepack is required" >&2; exit 1; }
dotnet --version
node --version
corepack --version
if [[ ! -f frontend/pnpm-lock.yaml ]]; then
echo "frontend/pnpm-lock.yaml is missing. Run 'cd frontend && corepack enable && pnpm install' and commit it." >&2
exit 2
fi