G7 외부 증거 3-artifact 동시 시간창 오케스트레이터

soak / runtime / topology 세 캡처는 같은 public host 의 겹치는 시간창이어야 하는데,
지금까지는 운영자가 세 명령을 따로 띄우고 시계를 손으로 맞춰야 했다. 50분짜리
실행에서 한 번 어긋나면 처음부터 다시 해야 하는 취약점이라 한 명령으로 묶었다.
세 캡처를 동시에 시작하고 전부 끝나면 human pack 을 더해 checker 까지 잇는다.

게이트를 약화시키지 않았다. CLI 로 실증한 fail-closed 경계 4종:
- 동의/장치 없이 운영 실행     physical_microphone_consent_required  exit 2
- human pack 없이 운영 실행    human_voice_gain_pack_required        exit 2
- 50분 미만 시간창             production_window_too_short           exit 2
- 세 캡처 host 불일치          hosts_must_match:[...]                exit 2

--rehearse 는 마이크를 열지 않고(soak 을 --preflight-only 로) 배관만 확인하며
보고서의 gate_closed 는 항상 false 다. 실제 실행 전 실패를 먼저 뽑기 위한 모드다.

기대 provider 기본값은 2026-08-08 결정에 맞춰 local_whisper / melotts 다.

이것으로 G7 에서 기계로 할 수 있는 부분은 끝났다. 남은 것은 코드로 만들 수 없는
둘뿐이다. 명시 동의 하 물리 마이크 50분 발화, 그리고 참가자 30명·독립 평가자 2인의
blind human voice-gain pack.

검증: 오케스트레이터 20/20, SSOT FAIL 0, dashboard E2E 10/10, ruff clean.
This commit is contained in:
Yun Chan 2026-08-08 09:40:08 +09:00
parent af993961c4
commit f3491be63f
5 changed files with 604 additions and 4 deletions

View file

@ -89,10 +89,17 @@
지원한다고 했으나 오답). Piper는 GPL, XTTS-v2·Fish Speech는 비상업이라 Higgs와 같은 문제다.
설치 함정 3가지는 `decisions/local-voice-stack.md`에 남겼다(setuptools<81, unidic download, eunjeon).
근거: [`decisions/local-voice-stack.md`](./decisions/local-voice-stack.md).
- [ ] **음성 캐스케이드 live** — Deepgram streaming adapter와 interim/final·word timestamp, bounded event queue,
운영 10MiB 사전 거부와 실제 PostgreSQL 동시 철회 직렬화는 코드/내부 회귀까지 완료했다. 종료에는 운영
key·quota를 사용한 live interim/final, 기대 provider/model과 일치하는 authenticated public WSS ready,
명시 동의 물리 마이크, 50분 양방향 장시간 실측과 실제 worker/proxy의 RSS·queue high-water가 모두 필요하다.
- [ ] **G7 외부 종료 게이트 — 남은 것은 사람 입력 둘뿐** — provider 의존은 해소됐다. 2026-08-08 로컬 스택
(faster-whisper STT + MeloTTS TTS, 둘 다 MIT)으로 바꿔 외부 STT 키가 더 이상 필요 없고, 체커의 벤더
하드코딩도 운영 provider 허용목록으로 교정했다. 기계적인 부분도
`scripts/run-g7-external-proof-window.py`가 soak·runtime·topology 세 캡처를 **하나의 겹치는 시간창**에서
동시에 돌리고 checker까지 이어서 실행하도록 묶었다(회귀 20/20).
**남은 것은 코드로 만들 수 없는 둘이다.**
(1) **명시 동의 하 물리 마이크 50분 양방향 soak** — 소유자가 실제로 50분 발화해야 한다.
(2) **독립 blind human voice-gain pack** — 참가자 30명 / 50회기 / 150 paired axis / 독립 평가자 2인 /
ICC(A,1) ≥ 0.75 · κ ≥ 0.70 · gain ≥ 0.01 · participant-cluster bootstrap 10,000회 95% CI lower > 0.
실행 전 `--rehearse`로 배관을 먼저 검증한다. fail-closed 경계 4종(동의 없음·pack 없음·50분 미만·host
불일치)은 CLI로 실증했다. 상세: `ops/outcome-os-g7-external-proof-readiness-2026-08-07.md`.
> **완료된 격리 검증 경로:** NAS 프리뷰 `http://100.116.83.60:8088`을 전용 Compose 프로젝트·포트·네트워크·볼륨에 배포하고 실제 브라우저 회기와 review API 저장 축어록을 확인했다. current source exact SHA `6030a677af7e87cbfabc422b553d108d53414fd3c446548734a13b036d35c611`은 전체 browser E2E 108/108, postdeploy SSE→DB review, health 3/3, OpenAPI 126·auth 401·G0~G8 route를 통과했다. 기존 프로젝트를 대상으로 한 중단·재생성 명령은 실행하지 않았다. 매일 04:30 KST `Vignette 회기 E2E 정기 검증`(automation id `vignette-e2e`)은 ACTIVE다. 새 기능이 material milestone이고 release-only patch 결정성·manifest checker·canonical clean-index·NAS preflight를 통과하며 배포 SHA가 달라질 때만 같은 격리 프리뷰를 갱신한다. 첫 스케줄 실행 이력은 아직 증거에 포함하지 않았다. 증거: [배포 증거](./ops/nas-preview-deployment-evidence-2026-08-07.md), [current 기계 판독 증거](./ops/evidence/nas-preview-current-deploy-2026-08-07.json), [브라우저 증거](./ops/evidence/nas-preview-live-turn-2026-08-07.png). 이는 G8 실제 receipt-bound rollback이나 미승인 물리 마이크·부재한 Deepgram 운영 키를 대체하지 않는다.
- [ ] **claude_cli ↔ Anthropic API live 동일성** — provider 라우팅·모델 탐색·설정 저장 경로는 구현 완료. 남은 게이트는 연구팀/기관 `ANTHROPIC_API_KEY`를 게이트웨이 호스트에 주입한 뒤 같은 프롬프트의 live 응답·계량·오류 표면화를 비교하는 것이다.

View file

@ -1122,6 +1122,7 @@
<tr><td>Deploy preflight</td><td><code>python scripts\check-deploy-preflight.py --skip-db --env-file infra\.env.example --allow-placeholder-secrets</code> / DB mode with local <code>DATABASE_URL</code></td><td>Passed: exact-pinned API requirements, live coaching <code>data/kb</code> source pack, env template keys, and DB readiness (<code>current_user=vignette</code>). DB mode can additionally check app-role DSN with <code>--require-app-role</code> and now verifies session read-model columns including <code>app.turns</code> voice metadata columns(<code>audio_ref</code>/<code>silence_ms</code>/<code>speech_rate</code>/<code>barge_in</code>/<code>provider_events</code>) plus worksheet review columns.</td></tr>
<tr><td>Fresh compose smoke</td><td><code>docker compose -p vignette-packaging-smoke -f infra/docker-compose.yml up -d --build</code> + proxy <code>/api/health</code></td><td>Passed with dummy production-safe env: API healthy, DB healthy, web/proxy up, <code>http://localhost:18080/api/health</code> 200 with <code>db:true</code>, <code>engine:true</code>, <code>engine_mode:"claude_cli"</code>. Smoke volumes/network removed after run.</td></tr>
<tr><td>격리 NAS 프리뷰 · 회기 E2E 자동화</td><td><code>http://100.116.83.60:8088</code> / <code>vignette-e2e</code> 매일 04:30 KST / <a href="./ops/nas-preview-deployment-evidence-2026-08-07.md">배포 증거</a> / <a href="./ops/evidence/nas-preview-current-deploy-2026-08-07.json">기계 판독 증거</a></td><td>current SHA <code>6030a677…c611</code>을 전용 Compose 프로젝트·포트·네트워크·named volume에 승격했다. browser E2E 108/108, health 3/3, OpenAPI 126·auth 401·G0~G8, postdeploy SSE→DB review를 통과했고 predeploy dump와 이전 exact images를 보존했다. 자동화는 ACTIVE이며 material milestone+release gate+배포 SHA 변경 때만 프리뷰를 갱신한다. G7 외부 mic/provider/human 증거는 별도다.</td></tr>
<tr><td>G7 외부 증거 3-artifact 동시 시간창 오케스트레이터</td><td><code>scripts/run-g7-external-proof-window.py</code> · <a href="./ops/outcome-os-g7-external-proof-readiness-2026-08-07.md">준비 문서</a></td><td>soak·runtime·topology 세 캡처는 같은 host의 겹치는 시간창이어야 하는데 지금까지는 운영자가 세 명령을 손으로 맞춰야 했다. 한 명령으로 동시에 띄우고 checker까지 잇도록 묶었다. 게이트는 약화시키지 않았다 — CLI 실증 fail-closed 4종: 동의·장치 없음 <code>physical_microphone_consent_required</code>, human pack 없음 <code>human_voice_gain_pack_required</code>, 50분 미만 <code>production_window_too_short</code>, host 불일치 <code>hosts_must_match</code>, 전부 exit 2. <code>--rehearse</code>는 마이크를 열지 않고 배관만 확인하며 <code>gate_closed</code>는 항상 false다. 회귀 <code>20/20</code>. 남은 것은 <b>물리 마이크 50분</b><b>독립 human voice-gain pack</b> 둘뿐이다.</td></tr>
<tr><td>노트북 로컬 음성 스택 — faster-whisper STT + MeloTTS TTS</td><td><code>scripts/local-whisper-stt-server.py</code> · <code>scripts/start-local-whisper-stt.ps1</code> · <code>voice_stt_provider=local_whisper</code></td><td>2026-08-08 소유자 결정으로 음성 양쪽을 노트북 상주 로컬 모델로 정했다. TTS는 처음 Higgs로 잡았으나 연구/비상업 라이선스라 <code>config.py</code>가 운영에서 차단하고 있었고 그 가드를 푸는 건 법적 판단이라, 상업 사용이 허용된 <b>MeloTTS Korean(MIT)</b>으로 바꿔 가드 자체를 불필요하게 만들었다. Kokoro-82M은 Apache 2.0이지만 공식 <code>VOICES.md</code><b>한국어가 없어</b> 탈락했다. 이전에는 로컬 STT 경로가 코드에 없었고 interim/final 스트리밍이 Deepgram 전용이었다. Higgs와 같은 loopback 사이드카를 추가해 오디오가 호스트를 벗어나지 않고 외부 STT 키도 필요 없다. <b>실측</b>: 저장소 합성 시드 8.72초에서 interim <code>9</code>·final <code>5</code>·word timestamp 12개 present, 전사는 원문과 한 글자 차이. 회귀 사이드카 <code>37/37</code>, API 전체 <code>908 passed</code>, gateway <code>58</code>, G7 checker <code>23</code>, ruff clean. 이 노트북은 CUDA는 보이지만 cuDNN 부재로 GPU 추론이 <b>네이티브 크래시</b>라 디바이스 확인을 자식 프로세스(<code>--self-check</code>)로 분리하고 CPU 폴백을 넣었다. cuDNN 9 설치는 소유자 판단으로 남겼다. MeloTTS는 CPU 정상 상태 RTF <code>0.27~0.28</code>(실시간 3.6배), <code>POST /tts</code> 200·WAV 350,566 bytes·3.61s, 빈 텍스트 422·미지 경로 404이며, <b>합성음을 로컬 STT가 완전 일치 전사</b>하는 왕복 검증을 통과했다(사이드카 16/16, API 914 passed). G7 체커의 <code>deepgram</code> 하드코딩도 운영 provider 허용목록으로 교정했고 선언/실제 불일치 차단은 유지한다. Higgs 운영 사용은 라이선스 가드가 남아 dev 전용이다.</td></tr>
<tr><td>비-secure origin 회기 리뷰 크래시 수정</td><td><code>apps/web/src/lib/uuid.ts</code> · <code>apps/web/e2e/insecure-context-uuid.spec.ts</code></td><td>격리 NAS 프리뷰(평문 HTTP·비-localhost)에서 회기 리뷰 라우트가 <code>crypto.randomUUID is not a function</code>으로 error boundary에 잡혔다. 이 API는 secure context 전용인데 제품 코드 18곳이 fallback 없이 호출했고 <code>RuptureRepairCard</code>는 렌더 시점 호출이라 화면 전체가 죽었다. 릴리스 게이트는 localhost 후보 스택에서만 돌아 미검출. <code>randomUuid()</code>로 통일하고 fallback도 <code>getRandomValues</code> 우선으로 예측 불가능성을 유지했다. 회귀 <code>6/6</code>(직접 호출 0건 검사 포함), typecheck·build 통과. <b>NAS 프리뷰에는 아직 미배포</b>이며 배포된 SHA <code>6030a677…c611</code>은 여전히 결함 빌드다.</td></tr>
<tr><td>공개 워치독 engine readiness 사각지대(해결)</td><td><code>scripts/watch-public-runtime.ps1</code> · <code>scripts/start-public-runtime.ps1</code> · <code>apps/api/engine_gateway/gateway.py</code></td><td>2026-08-07 22:0x KST 소유자 승인으로 9099를 재기동해 복구했다. 재기동 직후 <code>/ready?force=true</code> 200 <code>ok:true·detail:OK</code>, API 8001과 공개 <code>api-vignette.chanpaca.net</code> 모두 <code>status:ok·engine:true·engine_detail:OK</code>다. 원인은 16:06부터 상주한 9099 프로세스 하나였다(CLI·코드·cwd는 정상 — 동일 인자 직접 실행과 <code>EngineSession</code> 재현이 <code>"OK"</code>를 반환). 재발 방지 3건: (1) 워치독·기동 스크립트의 engine 판정을 <code>/health</code>(프로세스 liveness)에서 <code>/ready</code>(실제 생성)로 바꾸고 API 판정에 <code>engine</code>을 추가, shared secret 인스턴스용 토큰 헤더와 503 본문 로깅 포함. (2) 게이트웨이가 자식 <code>claude -p</code>의 stderr를 상시 드레인해(PIPE 미독 시 자식 블록도 차단) 실패 detail에 <code>exit</code>·stderr를 붙인다 — 실증: <code>empty engine response (error: unknown option '--vignette-nonexistent-flag-xyz')</code>. (3) 기동 시 게이트웨이 로그 회전, <code>Stop-UvicornByPort</code><code>Name -like python*</code> 추가(호출자 자기 자신 종료 방지). 검증: <code>pytest engine_gateway</code> 58/58, 워치독 <code>-CheckOnly</code> healthy 5/5, 장애 재현(9299·<code>CLAUDE_BIN</code> 부재)에서 <code>/health</code><code>ok:true</code>로 통과하고 <code>/ready</code>는 503으로 검출되어 워치독이 <code>unhealthy (1/3)</code> + 503 본문을 남겼다. <b>NAS용 9100(06:27 상주)은 실사용 세션 7개가 있어 미접촉</b>이라 게이트웨이 stderr 개선은 다음 재기동 때 반영된다.</td></tr>

View file

@ -62,3 +62,38 @@ G7은 계속 **BUILD**다. 이번 변경은 물리 마이크나 운영 Deepgram
네 파일을 `check-g7-external-proof.py`에 넣어 exit 0과 artifact SHA-256을 얻기 전에는 G7을
DONE으로 표시하지 않는다.
## 실행 오케스트레이터 (2026-08-08 추가)
앞의 1~3번은 **같은 public host의 겹치는 시간창**이어야 하는데, 지금까지는 운영자가 세 명령을 따로
띄우고 시계를 손으로 맞춰야 했다. 50분짜리 실행에서 한 번 어긋나면 처음부터 다시 해야 한다.
`scripts/run-g7-external-proof-window.py`가 세 캡처를 **동시에** 시작하고, 전부 끝나면 human pack을
더해 checker까지 그대로 돌린다.
```powershell
# 실제 실행 (물리 마이크 50분 + 사람 pack 필요)
& $py -X utf8 -B scripts/run-g7-external-proof-window.py `
--wss-url wss://api-vignette.chanpaca.net/voice/ws `
--origin https://api-vignette.chanpaca.net `
--admin-runtime-url https://api-vignette.chanpaca.net/admin/voice-runtime `
--compose-project <project> --api-container <c> --api-image-digest sha256:... `
--caddy-container <c> --caddy-image-digest sha256:... `
--microphone-device "<장치명>" --confirm-physical-capture `
--human-voice-gain <pack.json> --out-dir <증거디렉터리>
```
게이트를 **약화시키지 않는다.** CLI로 실증한 fail-closed 경계 네 가지:
| 시도 | 결과 |
|---|---|
| 동의·장치 없이 운영 실행 | `physical_microphone_consent_required`, exit 2 |
| human pack 없이 운영 실행 | `human_voice_gain_pack_required`, exit 2 |
| 50분 미만 시간창 | `production_window_too_short`, exit 2 |
| 세 캡처의 host 불일치 | `hosts_must_match:[...]`, exit 2 |
`--rehearse`는 마이크·사람 없이 **배관만** 확인하는 모드다. soak을 `--preflight-only`로 돌려 장치를
열지 않고, 인자·경로·산출 파일까지 실제로 검증한다. 보고서의 `gate_closed`는 rehearse에서 **항상
false**다. 실제 50분 실행 전에 이걸로 먼저 실패를 뽑아내라.
기대 provider 기본값은 2026-08-08 결정에 맞춰 `local_whisper` / `melotts`
(근거: `../decisions/local-voice-stack.md`). 회귀는 `scripts/test_run_g7_external_proof_window.py` 20/20.

View file

@ -0,0 +1,335 @@
#!/usr/bin/env python3
"""G7 외부 종료 증거 3종을 하나의 겹치는 시간창에서 동시에 잡는 오케스트레이터.
`check-g7-external-proof.py` artifact 요구하고, 개는 **같은 public host
겹치는 시간창**이어야 한다. 지금까지는 운영자가 명령을 따로 띄우고 시계를 손으로 맞춰야 했다.
50분짜리 실행에서 그건 어긋나면 처음부터 다시 해야 하는 종류의 취약점이다.
스크립트가 하는 :
1. soak / runtime / topology 캡처를 **동시에** 시작하고 전부 끝날 때까지 기다린다.
2. 캡처가 같은 public host 보는지 시작 전에 확인한다(fail-closed).
3. 끝나면 human voice-gain pack 더해 `check-g7-external-proof.py` 그대로 돌린다.
스크립트는 게이트를 **약화시키지 않는다**. 물리 마이크 캡처는 `--confirm-physical-capture`
`--microphone-device` 있어야 시작하고, human pack 만들어 주지 않는다. 번째 artifact
실제 참가자와 독립 평가자가 있어야 한다.
`--rehearse` 마이크·사람 없이 배관만 확인하는 모드다. 짧은 시간창으로 러너를 실제로 띄워
인자·경로·산출 파일까지 검증하되, soak `--preflight-only` 돌려 장치를 열지 않는다. 실제 50
실행 전에 이걸로 먼저 실패를 뽑아내라.
"""
from __future__ import annotations
import argparse
import json
import subprocess
import sys
from concurrent.futures import ThreadPoolExecutor
from dataclasses import dataclass
from pathlib import Path
from typing import Any, Callable, Iterable, Sequence
REPO_ROOT = Path(__file__).resolve().parents[1]
SCRIPTS = REPO_ROOT / "scripts"
SOAK_SCRIPT = SCRIPTS / "soak-public-voice-websocket.py"
RUNTIME_SCRIPT = SCRIPTS / "capture-g7-runtime-evidence.py"
TOPOLOGY_SCRIPT = SCRIPTS / "capture-g7-topology-evidence.py"
CHECKER_SCRIPT = SCRIPTS / "check-g7-external-proof.py"
MIN_PRODUCTION_SECONDS = 3_000.0
RUNTIME_SAMPLE_MARGIN = 1
class WindowError(RuntimeError):
"""Fail-closed 오케스트레이션 오류."""
@dataclass(frozen=True)
class Leg:
name: str
argv: list[str]
output: Path
@dataclass(frozen=True)
class LegResult:
name: str
returncode: int
output: Path
@property
def ok(self) -> bool:
return self.returncode == 0
def host_of(url: str) -> str:
from urllib.parse import urlsplit
return (urlsplit(url).hostname or "").lower()
def assert_single_host(*urls: str) -> str:
"""세 캡처가 같은 public host 를 보는지 확인한다."""
hosts = {host_of(url) for url in urls if url}
hosts.discard("")
if len(hosts) != 1:
raise WindowError(f"hosts_must_match:{sorted(hosts)}")
return hosts.pop()
def sample_plan(duration_seconds: float, interval_seconds: float) -> int:
"""시간창을 덮기에 충분한 샘플 수. 모자라면 겹침 검증이 깨진다."""
if duration_seconds <= 0 or interval_seconds <= 0:
raise WindowError("invalid_sampling_plan")
return int(duration_seconds // interval_seconds) + RUNTIME_SAMPLE_MARGIN
def build_soak_leg(args: argparse.Namespace, output: Path) -> Leg:
argv = [
sys.executable,
"-X",
"utf8",
str(SOAK_SCRIPT),
"--wss-url",
args.wss_url,
"--origin",
args.origin,
"--expected-stt-provider",
args.expected_stt_provider,
"--expected-stt-model",
args.expected_stt_model,
"--expected-tts-provider",
args.expected_tts_provider,
"--expected-tts-model",
args.expected_tts_model,
"--evidence-output",
str(output),
]
if args.rehearse:
# 배관 확인 전용. 장치를 열지 않는다.
argv.append("--preflight-only")
return Leg("voice_soak_preflight", argv, output)
if not args.microphone_device or not args.confirm_physical_capture:
raise WindowError("physical_microphone_consent_required")
argv += [
"--microphone-device",
args.microphone_device,
"--confirm-physical-capture",
"--duration-seconds",
str(args.duration_seconds),
]
return Leg("voice_soak", argv, output)
def build_runtime_leg(args: argparse.Namespace, output: Path) -> Leg:
samples = sample_plan(args.duration_seconds, args.runtime_interval_seconds)
argv = [
sys.executable,
"-X",
"utf8",
str(RUNTIME_SCRIPT),
"--url",
args.admin_runtime_url,
"--samples",
str(samples),
"--interval-seconds",
str(args.runtime_interval_seconds),
"--evidence-output",
str(output),
]
return Leg("runtime", argv, output)
def build_topology_leg(args: argparse.Namespace, output: Path) -> Leg:
samples = sample_plan(args.duration_seconds, args.topology_interval_seconds)
argv = [
sys.executable,
"-X",
"utf8",
str(TOPOLOGY_SCRIPT),
"--compose-project",
args.compose_project,
"--public-host",
host_of(args.wss_url),
"--api-container",
args.api_container,
"--api-image-digest",
args.api_image_digest,
"--caddy-container",
args.caddy_container,
"--caddy-image-digest",
args.caddy_image_digest,
"--samples",
str(samples),
"--interval-seconds",
str(args.topology_interval_seconds),
"--evidence-output",
str(output),
]
return Leg("topology", argv, output)
def plan_legs(args: argparse.Namespace, out_dir: Path) -> list[Leg]:
assert_single_host(args.wss_url, args.origin, args.admin_runtime_url)
return [
build_soak_leg(args, out_dir / "voice-soak.json"),
build_runtime_leg(args, out_dir / "runtime.json"),
build_topology_leg(args, out_dir / "topology.json"),
]
def _run(leg: Leg, *, timeout: float, runner: Callable[..., Any]) -> LegResult:
completed = runner(
leg.argv,
check=False,
capture_output=True,
text=True,
encoding="utf-8",
errors="replace",
timeout=timeout,
shell=False,
)
return LegResult(leg.name, int(completed.returncode), leg.output)
def run_window(
legs: Sequence[Leg],
*,
timeout: float,
runner: Callable[..., Any] | None = None,
) -> list[LegResult]:
"""세 캡처를 동시에 시작한다. 하나가 죽어도 나머지를 끝까지 돌린다."""
call = runner or subprocess.run
with ThreadPoolExecutor(max_workers=len(legs)) as pool:
futures = [
pool.submit(_run, leg, timeout=timeout, runner=call) for leg in legs
]
return [future.result() for future in futures]
def build_checker_argv(
results: Sequence[LegResult], human_pack: Path
) -> list[str]:
by_name = {result.name: result.output for result in results}
soak = by_name.get("voice_soak") or by_name.get("voice_soak_preflight")
if soak is None:
raise WindowError("soak_evidence_missing")
return [
sys.executable,
"-X",
"utf8",
str(CHECKER_SCRIPT),
"--voice-soak",
str(soak),
"--runtime",
str(by_name["runtime"]),
"--topology",
str(by_name["topology"]),
"--human-voice-gain",
str(human_pack),
"--json",
]
def summarize(
results: Sequence[LegResult],
*,
rehearse: bool,
checker_returncode: int | None,
) -> dict[str, Any]:
return {
"schema_version": "vignette.g7-external-proof-window.v1",
"mode": "rehearse" if rehearse else "production",
"legs": [
{"name": r.name, "returncode": r.returncode, "evidence": str(r.output)}
for r in results
],
"all_legs_passed": all(r.ok for r in results),
"checker_returncode": checker_returncode,
"gate_closed": checker_returncode == 0 and not rehearse,
"note": (
"rehearse 는 배관만 확인한다. 게이트는 실제 물리 마이크 50분 실행과 "
"독립 human voice-gain pack 이 있어야 닫힌다."
if rehearse
else "네 artifact 가 모두 통과해야 게이트가 닫힌다."
),
}
def build_parser() -> argparse.ArgumentParser:
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--wss-url", required=True)
parser.add_argument("--origin", required=True)
parser.add_argument("--admin-runtime-url", required=True)
parser.add_argument("--compose-project", required=True)
parser.add_argument("--api-container", required=True)
parser.add_argument("--api-image-digest", required=True)
parser.add_argument("--caddy-container", required=True)
parser.add_argument("--caddy-image-digest", required=True)
parser.add_argument("--expected-stt-provider", default="local_whisper")
parser.add_argument("--expected-stt-model", default="large-v3")
parser.add_argument("--expected-tts-provider", default="melotts")
parser.add_argument("--expected-tts-model", default="melotts-korean")
parser.add_argument("--microphone-device", default="")
parser.add_argument("--confirm-physical-capture", action="store_true")
parser.add_argument("--duration-seconds", type=float, default=MIN_PRODUCTION_SECONDS)
parser.add_argument("--runtime-interval-seconds", type=float, default=100.0)
parser.add_argument("--topology-interval-seconds", type=float, default=100.0)
parser.add_argument("--human-voice-gain", type=Path)
parser.add_argument("--out-dir", type=Path, required=True)
parser.add_argument(
"--rehearse",
action="store_true",
help="마이크 없이 배관만 확인한다. 게이트를 닫지 않는다.",
)
return parser
def validate(args: argparse.Namespace) -> None:
if not args.rehearse and args.duration_seconds < MIN_PRODUCTION_SECONDS:
raise WindowError("production_window_too_short")
if not args.rehearse and args.human_voice_gain is None:
raise WindowError("human_voice_gain_pack_required")
def main(argv: Iterable[str] | None = None) -> int:
args = build_parser().parse_args(list(argv) if argv is not None else None)
try:
validate(args)
args.out_dir.mkdir(parents=True, exist_ok=True)
legs = plan_legs(args, args.out_dir)
except WindowError as exc:
print(json.dumps({"ok": False, "error": str(exc)}, ensure_ascii=False), file=sys.stderr)
return 2
results = run_window(legs, timeout=args.duration_seconds + 900)
checker_returncode: int | None = None
if all(result.ok for result in results) and args.human_voice_gain is not None:
checker = subprocess.run(
build_checker_argv(results, args.human_voice_gain),
check=False,
capture_output=True,
text=True,
encoding="utf-8",
errors="replace",
shell=False,
)
checker_returncode = checker.returncode
report = summarize(
results, rehearse=args.rehearse, checker_returncode=checker_returncode
)
print(json.dumps(report, ensure_ascii=False, indent=2, sort_keys=True))
return 0 if report["all_legs_passed"] else 1
if __name__ == "__main__": # pragma: no cover - CLI
raise SystemExit(main())

View file

@ -0,0 +1,222 @@
from __future__ import annotations
import importlib.util
import sys
import unittest
from pathlib import Path
from types import SimpleNamespace
SCRIPT_PATH = Path(__file__).with_name("run-g7-external-proof-window.py")
SPEC = importlib.util.spec_from_file_location("run_g7_external_proof_window", SCRIPT_PATH)
assert SPEC is not None and SPEC.loader is not None
MODULE = importlib.util.module_from_spec(SPEC)
sys.modules[SPEC.name] = MODULE
SPEC.loader.exec_module(MODULE)
def args(**overrides):
base = {
"wss_url": "wss://api.example.test/voice/ws",
"origin": "https://api.example.test",
"admin_runtime_url": "https://api.example.test/admin/voice-runtime",
"compose_project": "vignette",
"api_container": "vignette-api-1",
"api_image_digest": "sha256:" + "a" * 64,
"caddy_container": "vignette-proxy-1",
"caddy_image_digest": "sha256:" + "b" * 64,
"expected_stt_provider": "local_whisper",
"expected_stt_model": "large-v3",
"expected_tts_provider": "melotts",
"expected_tts_model": "melotts-korean",
"microphone_device": "",
"confirm_physical_capture": False,
"duration_seconds": 3_000.0,
"runtime_interval_seconds": 100.0,
"topology_interval_seconds": 100.0,
"human_voice_gain": Path("pack.json"),
"out_dir": Path("out"),
"rehearse": False,
}
base.update(overrides)
return SimpleNamespace(**base)
class FakeCompleted:
def __init__(self, returncode: int) -> None:
self.returncode = returncode
self.stdout = ""
self.stderr = ""
class HostAlignmentTest(unittest.TestCase):
def test_matching_hosts_are_accepted(self) -> None:
host = MODULE.assert_single_host(
"wss://api.example.test/voice/ws",
"https://api.example.test",
"https://api.example.test/admin/voice-runtime",
)
self.assertEqual(host, "api.example.test")
def test_mismatched_hosts_fail_closed(self) -> None:
with self.assertRaises(MODULE.WindowError):
MODULE.assert_single_host(
"wss://api.example.test/voice/ws", "https://other.example.test"
)
def test_plan_rejects_a_split_window(self) -> None:
with self.assertRaises(MODULE.WindowError):
MODULE.plan_legs(
args(admin_runtime_url="https://elsewhere.test/admin/voice-runtime",
microphone_device="mic", confirm_physical_capture=True),
Path("out"),
)
class SamplePlanTest(unittest.TestCase):
def test_samples_cover_the_whole_window(self) -> None:
self.assertEqual(MODULE.sample_plan(3_000.0, 100.0), 31)
def test_invalid_plan_fails_closed(self) -> None:
for duration, interval in ((0, 100.0), (3_000.0, 0)):
with self.subTest(duration=duration, interval=interval):
with self.assertRaises(MODULE.WindowError):
MODULE.sample_plan(duration, interval)
class ConsentGateTest(unittest.TestCase):
def test_production_soak_requires_device_and_confirmation(self) -> None:
for device, confirm in (("", True), ("mic", False), ("", False)):
with self.subTest(device=device, confirm=confirm):
with self.assertRaises(MODULE.WindowError) as ctx:
MODULE.build_soak_leg(
args(microphone_device=device, confirm_physical_capture=confirm),
Path("soak.json"),
)
self.assertEqual(
str(ctx.exception), "physical_microphone_consent_required"
)
def test_confirmed_production_soak_passes_the_flags_through(self) -> None:
leg = MODULE.build_soak_leg(
args(microphone_device="mic-1", confirm_physical_capture=True),
Path("soak.json"),
)
self.assertEqual(leg.name, "voice_soak")
self.assertIn("--confirm-physical-capture", leg.argv)
self.assertEqual(leg.argv[leg.argv.index("--microphone-device") + 1], "mic-1")
self.assertNotIn("--preflight-only", leg.argv)
def test_rehearse_never_opens_a_microphone(self) -> None:
leg = MODULE.build_soak_leg(args(rehearse=True), Path("soak.json"))
self.assertEqual(leg.name, "voice_soak_preflight")
self.assertIn("--preflight-only", leg.argv)
self.assertNotIn("--confirm-physical-capture", leg.argv)
self.assertNotIn("--microphone-device", leg.argv)
def test_production_window_shorter_than_fifty_minutes_is_rejected(self) -> None:
with self.assertRaises(MODULE.WindowError) as ctx:
MODULE.validate(args(duration_seconds=600.0))
self.assertEqual(str(ctx.exception), "production_window_too_short")
def test_production_run_requires_the_human_pack(self) -> None:
with self.assertRaises(MODULE.WindowError) as ctx:
MODULE.validate(args(human_voice_gain=None))
self.assertEqual(str(ctx.exception), "human_voice_gain_pack_required")
def test_rehearse_relaxes_only_the_two_human_inputs(self) -> None:
MODULE.validate(args(rehearse=True, duration_seconds=30.0, human_voice_gain=None))
class LegCompositionTest(unittest.TestCase):
def test_expected_providers_default_to_the_decided_local_stack(self) -> None:
leg = MODULE.build_soak_leg(args(rehearse=True), Path("soak.json"))
self.assertEqual(
leg.argv[leg.argv.index("--expected-stt-provider") + 1], "local_whisper"
)
self.assertEqual(
leg.argv[leg.argv.index("--expected-tts-provider") + 1], "melotts"
)
def test_topology_leg_pins_the_public_host_from_the_wss_url(self) -> None:
leg = MODULE.build_topology_leg(args(), Path("topology.json"))
self.assertEqual(
leg.argv[leg.argv.index("--public-host") + 1], "api.example.test"
)
def test_all_three_legs_share_one_window_length(self) -> None:
legs = MODULE.plan_legs(
args(rehearse=True, duration_seconds=3_000.0), Path("out")
)
self.assertEqual([leg.name for leg in legs][1:], ["runtime", "topology"])
for leg in legs[1:]:
samples = int(leg.argv[leg.argv.index("--samples") + 1])
interval = float(leg.argv[leg.argv.index("--interval-seconds") + 1])
self.assertGreaterEqual(samples * interval, 3_000.0)
class WindowExecutionTest(unittest.TestCase):
def test_all_three_legs_start_before_any_finishes(self) -> None:
started: list[str] = []
def runner(argv, **kwargs):
started.append(argv[3])
return FakeCompleted(0)
legs = MODULE.plan_legs(args(rehearse=True), Path("out"))
results = MODULE.run_window(legs, timeout=60, runner=runner)
self.assertEqual(len(started), 3)
self.assertTrue(all(result.ok for result in results))
def test_one_failing_leg_does_not_cancel_the_others(self) -> None:
def runner(argv, **kwargs):
return FakeCompleted(1 if "capture-g7-runtime-evidence.py" in argv[3] else 0)
legs = MODULE.plan_legs(args(rehearse=True), Path("out"))
results = MODULE.run_window(legs, timeout=60, runner=runner)
self.assertEqual(len(results), 3)
failed = [r.name for r in results if not r.ok]
self.assertEqual(failed, ["runtime"])
class ReportTest(unittest.TestCase):
def _results(self, codes=(0, 0, 0)):
names = ("voice_soak", "runtime", "topology")
return [
MODULE.LegResult(name, code, Path(f"{name}.json"))
for name, code in zip(names, codes)
]
def test_rehearse_never_reports_the_gate_as_closed(self) -> None:
report = MODULE.summarize(
self._results(), rehearse=True, checker_returncode=0
)
self.assertFalse(report["gate_closed"])
self.assertEqual(report["mode"], "rehearse")
def test_production_gate_closes_only_on_checker_exit_zero(self) -> None:
closed = MODULE.summarize(
self._results(), rehearse=False, checker_returncode=0
)
open_gate = MODULE.summarize(
self._results(), rehearse=False, checker_returncode=1
)
self.assertTrue(closed["gate_closed"])
self.assertFalse(open_gate["gate_closed"])
def test_failed_leg_is_reported(self) -> None:
report = MODULE.summarize(
self._results((0, 1, 0)), rehearse=False, checker_returncode=None
)
self.assertFalse(report["all_legs_passed"])
self.assertFalse(report["gate_closed"])
def test_checker_argv_passes_all_four_artifacts(self) -> None:
argv = MODULE.build_checker_argv(self._results(), Path("pack.json"))
for flag in ("--voice-soak", "--runtime", "--topology", "--human-voice-gain"):
self.assertIn(flag, argv)
self.assertEqual(argv[argv.index("--human-voice-gain") + 1], "pack.json")
if __name__ == "__main__":
unittest.main()