tech-log-backend/fileserver-superpowers-package/validate_fileserver_docs.py

from __future__ import annotations

from collections import Counter
from pathlib import Path
import hashlib
import json
import re
import sys

ROOT = Path('/mnt/data')
DESIGN = ROOT / 'fileserver-platform-design.md'
PLAN = ROOT / 'fileserver-platform-implementation-plan.md'

errors: list[str] = []
checks: list[tuple[str, bool, str]] = []


def add(name: str, ok: bool, detail: str) -> None:
    checks.append((name, ok, detail))
    if not ok:
        errors.append(f'{name}: {detail}')


def sha256(path: Path) -> str:
    return hashlib.sha256(path.read_bytes()).hexdigest()

for path in (DESIGN, PLAN):
    add(f'{path.name} exists', path.exists(), str(path))

if errors:
    print('\n'.join(errors), file=sys.stderr)
    raise SystemExit(1)

design = DESIGN.read_text(encoding='utf-8')
plan = PLAN.read_text(encoding='utf-8')

add('design title', design.startswith('# Fileserver Platform 설계서'), True.__str__())
add('plan header', plan.startswith('# Fileserver Platform Implementation Plan\n\n> **For agentic workers:**'), 'required Superpowers header')
add('design code fences', design.count('```') % 2 == 0, f"count={design.count('```')}")
add('plan code fences', plan.count('```') % 2 == 0, f"count={plan.count('```')}")

for label, text in [('design', design), ('plan', plan)]:
    forbidden = [r'\bTBD\b', r'\bTODO\b', r'implement later', r'fill in details', r'Similar to Task']
    hits = [p for p in forbidden if re.search(p, text, re.I)]
    add(f'{label} placeholder scan', not hits, f'hits={hits}')

required_design_sections = [
    '## 5. 지원 매트릭스',
    '## 6. 전체 아키텍처',
    '## 9. 상태 머신과 invariant',
    '## 10. Metadata Store 설계',
    '## 11. Content Store Port',
    '## 12. Local Filesystem Adapter',
    '## 14. Publish와 완료 처리',
    '## 15. Upload Application 설계',
    '## 19. HTTP API',
    '## 20. Range와 Conditional Request',
    '## 21. Spring MVC Adapter',
    '## 22. Spring WebFlux Adapter',
    '## 23. Nginx 전송 위임',
    '## 24. 재개 가능한 업로드',
    '## 27. 보안 정책',
    '## 28. 다중 인스턴스와 NFS',
    '## 30. 관측성',
    '## 33. 테스트 전략',
    '## 37. 완료 정의',
]
missing_sections = [s for s in required_design_sections if s not in design]
add('design section coverage', not missing_sections, f'missing={missing_sections}')

source_topics = {
    'MVC': ['Spring MVC Adapter', 'MvcTransferExecutorProperties'],
    'WebFlux': ['Spring WebFlux Adapter', 'DataBuffer'],
    'local/PVC/NFS': ['Kubernetes PVC', 'NFSv4.1', 'Local Filesystem Adapter'],
    'content/metadata separation': ['Content Store Port', 'Metadata Store 설계'],
    'upload': ['Upload Application 설계', 'multipart', 'application/octet-stream'],
    'download': ['Range와 Conditional Request', 'ETag', 'If-Range'],
    'publish': ['ATOMIC_MOVE_REQUIRED', 'METADATA_POINTER', 'AmbiguousCompletionException'],
    'security': ['traversal', 'symlink', 'READY gate'],
    'resumable': ['tus 1.0 Stable', 'draft-12 Experimental'],
    'observability': ['Metric', 'Trace', 'Audit'],
}
for topic, needles in source_topics.items():
    missing = [n for n in needles if n not in design]
    add(f'design topic: {topic}', not missing, f'missing={missing}')

# Core Port snippet must not expose adapter types.
port_match = re.search(r'### 11\.2 Blocking SPI\n(.*?)### 11\.3 Async SPI', design, re.S)
port_text = port_match.group(1) if port_match else ''
forbidden_port_types = ['java.nio.file.Path', 'org.springframework.core.io.Resource', 'DataBuffer', 'Flux<']
port_hits = [x for x in forbidden_port_types if x in port_text]
add('blocking core port leakage', bool(port_match) and not port_hits, f'hits={port_hits}')

# Task structure.
task_matches = list(re.finditer(r'^### Task (\d+):', plan, re.M))
task_numbers = [int(m.group(1)) for m in task_matches]
add('task count', len(task_numbers) == 33, f'count={len(task_numbers)}')
add('task numbering', task_numbers == list(range(1, 34)), f'numbers={task_numbers}')

missing_task_blocks: dict[int, list[str]] = {}
for idx, match in enumerate(task_matches):
    end = task_matches[idx + 1].start() if idx + 1 < len(task_matches) else plan.find('\n## 3.', match.start())
    segment = plan[match.start():end]
    required = [
        '**Files:**', '**Interfaces:**', '**Step 1:', '**Step 2:',
        '**Step 3:', '**Step 4:', '**Step 5:', 'Expected:', 'git commit'
    ]
    missing = [item for item in required if item not in segment]
    if missing:
        missing_task_blocks[int(match.group(1))] = missing
add('task block completeness', not missing_task_blocks, json.dumps(missing_task_blocks, ensure_ascii=False))

create_paths = re.findall(r'^- Create: `([^`]+)`', plan, re.M)
duplicates = {path: count for path, count in Counter(create_paths).items() if count > 1}
add('unique create paths', not duplicates, json.dumps(duplicates, ensure_ascii=False))

required_plan_topics = [
    'Task 10: Storage capability probe',
    'Task 11: Streaming append',
    'Task 13: Atomic move와 metadata pointer publish',
    'Task 18: HTTP Range',
    'Task 21: Spring WebFlux raw·multipart upload',
    'Task 23: Nginx `X-Accel-Redirect`',
    'Task 26: 다중 인스턴스 writer lease',
    'Task 27: tus 1.0 Stable',
    'Task 28: HTTPbis resumable upload draft-12 Experimental',
    'Task 29: HTTP Problem Detail과 보안 hardening',
    'Task 32: Filesystem, HTTP, fault, performance Testkit',
    'Task 33: CI matrix',
]
missing_plan_topics = [x for x in required_plan_topics if x not in plan]
add('plan scope coverage', not missing_plan_topics, f'missing={missing_plan_topics}')

add('no Redis carryover', 'redis' not in design.lower() and 'redis' not in plan.lower(), 'search term=redis')
add('no deprecated nginx token design', 'DelegatedPathToken' not in design + plan and 'opaque-token' not in design + plan, 'token mapper removed')

status = 'PASS' if not errors else 'FAIL'
report = ROOT / 'fileserver-superpowers-validation.md'
lines = [
    '# Fileserver Superpowers 문서 검증',
    '',
    f'**결과:** {status}',
    '',
    '## 파일',
    '',
    f'- `{DESIGN.name}` — {len(design.splitlines())} lines, {len(design.encode())} bytes, SHA-256 `{sha256(DESIGN)}`',
    f'- `{PLAN.name}` — {len(plan.splitlines())} lines, {len(plan.encode())} bytes, SHA-256 `{sha256(PLAN)}`',
    '',
    '## 검증 항목',
    '',
]
for name, ok, detail in checks:
    lines.append(f"- [{'x' if ok else ' '}] **{name}** — {detail}")

lines += [
    '',
    '## 검증 범위의 한계',
    '',
    '- 현재 Backend Skeleton 저장소가 입력되지 않아 Gradle compilation, integration test, Nginx execution, PVC·NFS certification은 실행하지 않았다.',
    '- 본 검증은 설계·계획 문서의 구조, 내부 일관성, 범위 추적성, 미확정 표식과 중복 경로를 확인한 정적 검증이다.',
]
report.write_text('\n'.join(lines) + '\n', encoding='utf-8')

print(json.dumps({
    'status': status,
    'errors': errors,
    'checks': len(checks),
    'design_lines': len(design.splitlines()),
    'plan_lines': len(plan.splitlines()),
    'task_count': len(task_numbers),
    'report': str(report),
}, ensure_ascii=False, indent=2))

raise SystemExit(0 if not errors else 1)