16e791e044
8월 7일까지 워킹트리에만 남아 있던 미커밋 작업을 커밋한다. 여러 사본 폴더(worktree·clone)에 흩어져 있던 중간 스냅샷을 정리하기 전에 원본을 git 이력으로 고정하는 것이 목적이다. - contracts/routes/services: measurement, outcome_trajectory, rupture_repair, deliberate_practice, calibration_transfer, supervision_research, multimodal_alliance, continuous_improvement 계열 신규 모듈과 테스트 - infra/db/init: 07~16 마이그레이션(측정 기반~calibration transfer 실행) - apps/web: 세션 리뷰 카드·관리 화면·E2E 스펙 추가 - docs/ops: G0~G8 라이브 통합·배포·롤백 증거 문서와 evidence JSON/PNG - scripts: smoke·ledger·릴리스 에이전트·NAS 프리뷰 운영 스크립트 engine.public 로그 .bak과 apps/web/test-results 산출물은 커밋에서 제외했다.
621 lines
22 KiB
Python
621 lines
22 KiB
Python
"""Exercise the G7 multimodal-alliance lifecycle over live HTTP.
|
|
|
|
The smoke uses a synthetic WAV only to derive deterministic metadata. It never
|
|
uploads or prints transcript/audio content. Durable fixtures are intentionally
|
|
left in the development database as auditable integration evidence.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import copy
|
|
import hashlib
|
|
import io
|
|
import json
|
|
import math
|
|
import secrets
|
|
import struct
|
|
import time
|
|
import urllib.error
|
|
import urllib.request
|
|
import wave
|
|
from dataclasses import dataclass
|
|
from datetime import UTC, datetime
|
|
from http.cookiejar import CookieJar
|
|
from pathlib import Path
|
|
from typing import Any
|
|
from uuid import uuid4
|
|
|
|
|
|
COHORT_ID = "e2e-hanshin"
|
|
OTHER_COHORT_ID = "e2e-other-cohort"
|
|
INTERNAL_HEADER = "X-Vignette-Multimodal-Token"
|
|
|
|
|
|
class SmokeError(RuntimeError):
|
|
pass
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class ApiResponse:
|
|
status: int
|
|
body: Any
|
|
|
|
|
|
class ApiClient:
|
|
def __init__(self, base_url: str, timeout: float) -> None:
|
|
self.base_url = base_url.rstrip("/")
|
|
self.timeout = timeout
|
|
self._opener = urllib.request.build_opener(
|
|
urllib.request.HTTPCookieProcessor(CookieJar())
|
|
)
|
|
|
|
def request(
|
|
self,
|
|
method: str,
|
|
path: str,
|
|
payload: dict[str, Any] | None = None,
|
|
*,
|
|
expected: set[int] | None = None,
|
|
headers: dict[str, str] | None = None,
|
|
) -> ApiResponse:
|
|
data = None
|
|
request_headers = {"Accept": "application/json", **(headers or {})}
|
|
if payload is not None:
|
|
data = json.dumps(payload, ensure_ascii=False).encode("utf-8")
|
|
request_headers["Content-Type"] = "application/json"
|
|
request = urllib.request.Request(
|
|
f"{self.base_url}{path}",
|
|
data=data,
|
|
headers=request_headers,
|
|
method=method,
|
|
)
|
|
try:
|
|
with self._opener.open(request, timeout=self.timeout) as response:
|
|
raw = response.read().decode("utf-8")
|
|
result = ApiResponse(response.status, json.loads(raw) if raw else {})
|
|
except urllib.error.HTTPError as exc:
|
|
raw = exc.read().decode("utf-8", errors="replace")
|
|
try:
|
|
body = json.loads(raw) if raw else {}
|
|
except json.JSONDecodeError:
|
|
body = {"detail": raw[:500]}
|
|
result = ApiResponse(exc.code, body)
|
|
except urllib.error.URLError as exc:
|
|
raise SmokeError(
|
|
f"{method} {path} transport failed: {type(exc.reason).__name__}"
|
|
) from exc
|
|
if result.status not in (expected or {200}):
|
|
detail = result.body.get("detail") if isinstance(result.body, dict) else None
|
|
raise SmokeError(
|
|
f"{method} {path} returned HTTP {result.status}; detail={detail!r}"
|
|
)
|
|
return result
|
|
|
|
|
|
def _sign_in(
|
|
client: ApiClient,
|
|
*,
|
|
suffix: str,
|
|
identity: str,
|
|
role: str,
|
|
cohort_ids: list[str],
|
|
) -> str:
|
|
login = client.request(
|
|
"POST",
|
|
"/auth/dev-login",
|
|
{
|
|
"email": f"dev.e2e.multimodal.{identity}.{suffix}@hs.ac.kr",
|
|
"role": role,
|
|
"display_name": f"Multimodal {identity.title()}",
|
|
"cohort_ids": cohort_ids,
|
|
},
|
|
)
|
|
client.request(
|
|
"POST",
|
|
"/users/me/onboarding",
|
|
{
|
|
"legal_name": f"Multimodal {identity.title()}",
|
|
"affiliation": "한신대학교",
|
|
"department": "상담심리학과",
|
|
"grade_level": "통합검증",
|
|
"phone": "010-0000-0000",
|
|
"contact_address": "경기도 오산시 한신대학교",
|
|
"nickname": f"Multimodal {identity.title()}",
|
|
"self_introduction": "G7 멀티모달 동맹 API 검증 fixture입니다.",
|
|
"avatar_url": "",
|
|
"terms_accepted": True,
|
|
"privacy_accepted": True,
|
|
},
|
|
)
|
|
user_id = str(login.body.get("user_id") or "")
|
|
if not user_id:
|
|
raise SmokeError(f"dev-login omitted user_id for {identity}")
|
|
return user_id
|
|
|
|
|
|
def _choose_persona(client: ApiClient) -> str:
|
|
response = client.request("GET", "/personas")
|
|
usable = [
|
|
item
|
|
for item in response.body
|
|
if isinstance(item, dict)
|
|
and item.get("source") == "database"
|
|
and not item.get("degraded")
|
|
and item.get("code")
|
|
]
|
|
if not usable:
|
|
raise SmokeError("persona catalog has no non-degraded database persona")
|
|
preferred = next((item for item in usable if item.get("code") == "P1"), usable[0])
|
|
return str(preferred["code"])
|
|
|
|
|
|
def _synthetic_wav() -> bytes:
|
|
sample_rate = 16_000
|
|
frames = sample_rate * 1200 // 1000
|
|
output = io.BytesIO()
|
|
with wave.open(output, "wb") as writer:
|
|
writer.setnchannels(1)
|
|
writer.setsampwidth(2)
|
|
writer.setframerate(sample_rate)
|
|
payload = bytearray()
|
|
for index in range(frames):
|
|
value = int(3200 * math.sin(2 * math.pi * 220 * index / sample_rate))
|
|
payload.extend(struct.pack("<h", value))
|
|
writer.writeframes(bytes(payload))
|
|
return output.getvalue()
|
|
|
|
|
|
def _timeline(live_suffix: str) -> dict[str, Any]:
|
|
return {
|
|
"audio_duration_ms": 1200,
|
|
"words": [
|
|
{
|
|
"word_index": 0,
|
|
"start_ms": 40,
|
|
"end_ms": 220,
|
|
"speaker": "learner",
|
|
"token_hash": "a" * 64,
|
|
},
|
|
{
|
|
"word_index": 1,
|
|
"start_ms": 500,
|
|
"end_ms": 710,
|
|
"speaker": "client",
|
|
"token_hash": "b" * 64,
|
|
},
|
|
],
|
|
"events": [
|
|
{
|
|
"event_id": f"oas-g7-event-{live_suffix}-silence",
|
|
"event_type": "silence",
|
|
"start_ms": 220,
|
|
"end_ms": 500,
|
|
"actor": "both",
|
|
"observed_feature": "280ms turn transition silence",
|
|
"uncertainty": 0.05,
|
|
"source": "stt_word_timestamps",
|
|
},
|
|
{
|
|
"event_id": f"oas-g7-event-{live_suffix}-overlap",
|
|
"event_type": "overlap",
|
|
"start_ms": 710,
|
|
"end_ms": 770,
|
|
"actor": "both",
|
|
"observed_feature": "60ms simultaneous speech segment",
|
|
"uncertainty": 0.1,
|
|
"source": "observed_audio_runtime",
|
|
},
|
|
{
|
|
"event_id": f"oas-g7-event-{live_suffix}-interruption",
|
|
"event_type": "interruption",
|
|
"start_ms": 770,
|
|
"end_ms": 840,
|
|
"actor": "learner",
|
|
"observed_feature": "learner segment began before client segment ended",
|
|
"uncertainty": 0.12,
|
|
"source": "observed_audio_runtime",
|
|
},
|
|
{
|
|
"event_id": f"oas-g7-event-{live_suffix}-prosody",
|
|
"event_type": "prosody",
|
|
"start_ms": 840,
|
|
"end_ms": 1080,
|
|
"actor": "learner",
|
|
"observed_feature": "median intensity decreased by 3dB",
|
|
"uncertainty": 0.2,
|
|
"source": "observed_audio_runtime",
|
|
},
|
|
],
|
|
}
|
|
|
|
|
|
def _stable_retry(
|
|
client: ApiClient,
|
|
path: str,
|
|
body: dict[str, Any],
|
|
*,
|
|
id_field: str,
|
|
headers: dict[str, str] | None = None,
|
|
) -> dict[str, Any]:
|
|
created = client.request("POST", path, body, expected={201}, headers=headers)
|
|
retried = client.request("POST", path, body, expected={201}, headers=headers)
|
|
if created.body.get(id_field) != retried.body.get(id_field):
|
|
raise SmokeError(f"same submission changed {id_field} for {path}")
|
|
if retried.body.get("idempotent_replay") is not True:
|
|
raise SmokeError(f"same submission retry was not stable for {path}")
|
|
return created.body
|
|
|
|
|
|
def _assert_safe_metadata(value: Any, path: str = "response") -> None:
|
|
if isinstance(value, dict):
|
|
forbidden = {
|
|
"raw_transcript",
|
|
"transcript",
|
|
"utterance_text",
|
|
"diagnosis",
|
|
"emotion_label",
|
|
"total_score",
|
|
} & set(value)
|
|
if forbidden:
|
|
raise SmokeError(f"{path} exposed forbidden keys: {sorted(forbidden)}")
|
|
if value.get("clinical_claim_allowed") is True:
|
|
raise SmokeError(f"{path} enabled a clinical claim")
|
|
for key, child in value.items():
|
|
_assert_safe_metadata(child, f"{path}.{key}")
|
|
elif isinstance(value, list):
|
|
for index, child in enumerate(value):
|
|
_assert_safe_metadata(child, f"{path}[{index}]")
|
|
|
|
|
|
def _prepare_case(case: dict[str, Any], suffix: str) -> dict[str, Any]:
|
|
result = copy.deepcopy(case)
|
|
for key in ("text_measurement", "voice_measurement"):
|
|
measurement = result[key]
|
|
measurement["measurement_id"] = f"{measurement['measurement_id']}-{suffix}"
|
|
if measurement["status"] == "ready":
|
|
measurement["model_run_id"] = str(uuid4())
|
|
result["calibration"]["calibration_id"] = (
|
|
f"{result['calibration']['calibration_id']}-{suffix}"
|
|
)
|
|
return result
|
|
|
|
|
|
def run(args: argparse.Namespace) -> dict[str, Any]:
|
|
if len(args.internal_token) < 32:
|
|
raise SmokeError("--internal-token must contain at least 32 characters")
|
|
root = ApiClient(args.api_base_url, args.request_timeout)
|
|
health = root.request("GET", "/health")
|
|
if not health.body.get("db") or not health.body.get("engine"):
|
|
raise SmokeError("API health is not DB+engine ready")
|
|
|
|
internal = ApiClient(args.api_base_url, args.request_timeout)
|
|
internal.request(
|
|
"POST",
|
|
"/internal/sessions/00000000-0000-4000-8000-000000000000/multimodal-alliance/timelines",
|
|
{},
|
|
expected={401},
|
|
)
|
|
internal.request(
|
|
"POST",
|
|
"/internal/sessions/00000000-0000-4000-8000-000000000000/multimodal-alliance/timelines",
|
|
{},
|
|
expected={403},
|
|
headers={INTERNAL_HEADER: "wrong-token"},
|
|
)
|
|
token_header = {INTERNAL_HEADER: args.internal_token}
|
|
|
|
login_suffix = f"{int(time.time())}.{secrets.token_hex(3)}"
|
|
live_suffix = secrets.token_hex(5)
|
|
learner = ApiClient(args.api_base_url, args.request_timeout)
|
|
teacher = ApiClient(args.api_base_url, args.request_timeout)
|
|
outsider = ApiClient(args.api_base_url, args.request_timeout)
|
|
learner_id = _sign_in(
|
|
learner,
|
|
suffix=login_suffix,
|
|
identity="learner",
|
|
role="learner",
|
|
cohort_ids=[COHORT_ID],
|
|
)
|
|
_sign_in(
|
|
teacher,
|
|
suffix=login_suffix,
|
|
identity="teacher",
|
|
role="teacher",
|
|
cohort_ids=[COHORT_ID],
|
|
)
|
|
_sign_in(
|
|
outsider,
|
|
suffix=login_suffix,
|
|
identity="outsider",
|
|
role="teacher",
|
|
cohort_ids=[OTHER_COHORT_ID],
|
|
)
|
|
|
|
started = learner.request(
|
|
"POST",
|
|
"/sessions",
|
|
{
|
|
"persona_code": _choose_persona(learner),
|
|
"theory_mode": "humanistic",
|
|
"goal_stages": ["라포", "탐색"],
|
|
},
|
|
expected={201},
|
|
)
|
|
session_id = str(started.body.get("session_id") or "")
|
|
if not session_id or started.body.get("degraded"):
|
|
raise SmokeError("session start did not produce a durable non-degraded session")
|
|
|
|
consent_path = f"/sessions/{session_id}/multimodal-alliance/consent"
|
|
consent_body = {
|
|
"submission_id": str(uuid4()),
|
|
"consent_status": "granted",
|
|
"retain_audio": True,
|
|
"retain_derived_features": True,
|
|
"transcript_retained": True,
|
|
"retention_days": 30,
|
|
"policy_version": "g7-live-v1",
|
|
"reason_code": None,
|
|
}
|
|
consent = _stable_retry(
|
|
learner,
|
|
consent_path,
|
|
consent_body,
|
|
id_field="consent_snapshot_id",
|
|
)
|
|
changed_consent = copy.deepcopy(consent_body)
|
|
changed_consent["retention_days"] = 31
|
|
learner.request("POST", consent_path, changed_consent, expected={409})
|
|
|
|
synthetic_audio = _synthetic_wav()
|
|
timeline_body = {
|
|
"submission_id": str(uuid4()),
|
|
"timeline": _timeline(live_suffix),
|
|
"audio_asset": {
|
|
"audio_ref": f"g7-live://synthetic/{live_suffix}.wav",
|
|
"audio_sha256": hashlib.sha256(synthetic_audio).hexdigest(),
|
|
"media_type": "audio/wav",
|
|
"byte_size": len(synthetic_audio),
|
|
},
|
|
}
|
|
timeline_path = f"/internal/sessions/{session_id}/multimodal-alliance/timelines"
|
|
timeline = _stable_retry(
|
|
internal,
|
|
timeline_path,
|
|
timeline_body,
|
|
id_field="timeline_id",
|
|
headers=token_header,
|
|
)
|
|
changed_timeline = copy.deepcopy(timeline_body)
|
|
changed_timeline["timeline"]["audio_duration_ms"] = 1300
|
|
internal.request(
|
|
"POST", timeline_path, changed_timeline, expected={409}, headers=token_header
|
|
)
|
|
|
|
pack = json.loads(Path(args.benchmark_path).read_text(encoding="utf-8"))
|
|
measurement_path = (
|
|
f"/internal/sessions/{session_id}/multimodal-alliance/measurements"
|
|
)
|
|
fusion_results: list[dict[str, Any]] = []
|
|
first_body: dict[str, Any] | None = None
|
|
provenance = {
|
|
"instrument_id": "alliance-axis-observer",
|
|
"instrument_version": "g7-live-v1",
|
|
"model_name": "synthetic-observer",
|
|
"prompt_version": "g7-live-v1",
|
|
}
|
|
for index, source_case in enumerate(pack["cases"]):
|
|
case = _prepare_case(source_case, live_suffix)
|
|
body = {
|
|
"submission_id": str(uuid4()),
|
|
"text_measurement": case["text_measurement"],
|
|
"text_provenance": provenance,
|
|
"voice_measurement": case["voice_measurement"],
|
|
"voice_provenance": provenance,
|
|
"calibration": case["calibration"],
|
|
}
|
|
if index == 0:
|
|
result = _stable_retry(
|
|
internal,
|
|
measurement_path,
|
|
body,
|
|
id_field="fusion_record_id",
|
|
headers=token_header,
|
|
)
|
|
first_body = body
|
|
else:
|
|
result = internal.request(
|
|
"POST", measurement_path, body, expected={201}, headers=token_header
|
|
).body
|
|
applied = bool(result["result"]["fusion_applied"])
|
|
if applied is not bool(case["expected_fusion_applied"]):
|
|
raise SmokeError(f"fusion gate mismatch for {case['case_id']}")
|
|
fusion_results.append(result)
|
|
if first_body is None:
|
|
raise SmokeError("benchmark pack did not yield a measurement case")
|
|
changed_measurement = copy.deepcopy(first_body)
|
|
changed_measurement["text_measurement"]["value"] = 0.63
|
|
internal.request(
|
|
"POST",
|
|
measurement_path,
|
|
changed_measurement,
|
|
expected={409},
|
|
headers=token_header,
|
|
)
|
|
|
|
metadata_path = f"/sessions/{session_id}/multimodal-alliance"
|
|
learner_view = learner.request("GET", metadata_path).body
|
|
teacher_view = teacher.request("GET", metadata_path).body
|
|
outsider.request("GET", metadata_path, expected={404})
|
|
teacher.request(
|
|
"GET", f"{metadata_path}/raw-audio", expected={403}
|
|
)
|
|
raw_before = learner.request("GET", f"{metadata_path}/raw-audio").body
|
|
_assert_safe_metadata(learner_view)
|
|
_assert_safe_metadata(teacher_view)
|
|
if len(learner_view.get("word_timestamps") or []) != 2:
|
|
raise SmokeError("single audio clock did not retain two word timestamps")
|
|
if len(learner_view.get("voice_events") or []) != 4:
|
|
raise SmokeError("single audio clock did not retain four interaction events")
|
|
if len(learner_view.get("measurements") or []) != 6:
|
|
raise SmokeError("text and voice measurements were not independently retained")
|
|
if len(learner_view.get("fusion_decisions") or []) != 3:
|
|
raise SmokeError("all three benchmark fusion decisions were not retained")
|
|
if len(raw_before.get("items") or []) != 1:
|
|
raise SmokeError("learner raw-audio access did not expose one retained asset")
|
|
|
|
deletion_body = {
|
|
"submission_id": str(uuid4()),
|
|
"scopes": ["audio", "derived_features"],
|
|
}
|
|
deletion_path = f"{metadata_path}/deletion-requests"
|
|
explicit_deletion = _stable_retry(
|
|
learner,
|
|
deletion_path,
|
|
deletion_body,
|
|
id_field="deletion_request_id",
|
|
)
|
|
|
|
withdrawal_body = {
|
|
"submission_id": str(uuid4()),
|
|
"policy_version": "g7-live-v1",
|
|
"reason_code": "learner_withdrawal",
|
|
"transcript_retained": True,
|
|
}
|
|
withdrawal = _stable_retry(
|
|
learner,
|
|
f"/sessions/{session_id}/multimodal-alliance/withdraw",
|
|
withdrawal_body,
|
|
id_field="consent_snapshot_id",
|
|
)
|
|
internal.request(
|
|
"POST",
|
|
timeline_path,
|
|
{
|
|
**timeline_body,
|
|
"submission_id": str(uuid4()),
|
|
},
|
|
expected={409},
|
|
headers=token_header,
|
|
)
|
|
|
|
deleted_at = datetime.now(UTC).isoformat().replace("+00:00", "Z")
|
|
tombstones = [
|
|
{
|
|
"scope": "audio",
|
|
"target_ref_hash": hashlib.sha256(
|
|
timeline_body["audio_asset"]["audio_ref"].encode("utf-8")
|
|
).hexdigest(),
|
|
"deletion_proof": "synthetic object-store delete acknowledged",
|
|
"deleted_at": deleted_at,
|
|
},
|
|
{
|
|
"scope": "derived_features",
|
|
"target_ref_hash": hashlib.sha256(
|
|
str(timeline["timeline_id"]).encode("utf-8")
|
|
).hexdigest(),
|
|
"deletion_proof": "derived feature access tombstoned",
|
|
"deleted_at": deleted_at,
|
|
},
|
|
]
|
|
completion_ids: list[str] = []
|
|
for deletion_request_id in (
|
|
explicit_deletion["deletion_request_id"],
|
|
withdrawal["deletion_request_id"],
|
|
):
|
|
completion = _stable_retry(
|
|
internal,
|
|
f"/internal/multimodal-alliance/deletion-requests/{deletion_request_id}/complete",
|
|
{
|
|
"submission_id": str(uuid4()),
|
|
"actor_uid": None,
|
|
"actor_kind": "retention_worker",
|
|
"tombstones": tombstones,
|
|
},
|
|
id_field="deletion_request_id",
|
|
headers=token_header,
|
|
)
|
|
completion_ids.extend(str(item) for item in completion["tombstone_ids"])
|
|
|
|
after = learner.request("GET", metadata_path).body
|
|
raw_after = learner.request("GET", f"{metadata_path}/raw-audio").body
|
|
if raw_after.get("items"):
|
|
raise SmokeError("audio tombstone did not remove raw-audio access")
|
|
if after.get("word_timestamps") or after.get("voice_events"):
|
|
raise SmokeError("derived-feature tombstone did not hide voice timeline detail")
|
|
if any(item.get("modality") != "text" for item in after.get("measurements") or []):
|
|
raise SmokeError("deleted derived features leaked voice measurements")
|
|
if any(
|
|
"voice" in (item.get("modalities_used") or [])
|
|
for item in after.get("fusion_decisions") or []
|
|
):
|
|
raise SmokeError("deleted derived features leaked voice fusion decisions")
|
|
|
|
return {
|
|
"schema": "vignette.multimodal-alliance-live-api-evidence.v1",
|
|
"generated_at": datetime.now(UTC).isoformat(),
|
|
"api_base_url": args.api_base_url.rstrip("/"),
|
|
"health": {"db": True, "engine": True},
|
|
"learner_id": learner_id,
|
|
"session_id": session_id,
|
|
"consent_snapshot_id": str(consent["consent_snapshot_id"]),
|
|
"timeline_id": str(timeline["timeline_id"]),
|
|
"audio_asset_id": str(timeline["audio_asset_id"]),
|
|
"fusion_record_ids": [str(item["fusion_record_id"]) for item in fusion_results],
|
|
"fusion_applied": [bool(item["result"]["fusion_applied"]) for item in fusion_results],
|
|
"withdrawal_snapshot_id": str(withdrawal["consent_snapshot_id"]),
|
|
"deletion_request_ids": [
|
|
str(explicit_deletion["deletion_request_id"]),
|
|
str(withdrawal["deletion_request_id"]),
|
|
],
|
|
"tombstone_ids": completion_ids,
|
|
"assertions": {
|
|
"internal_missing_token_401": True,
|
|
"internal_wrong_token_403": True,
|
|
"stable_retries": True,
|
|
"changed_payload_conflicts_409": True,
|
|
"single_clock_word_count": 2,
|
|
"interaction_event_count": 4,
|
|
"independent_measurement_count": 6,
|
|
"fusion_benchmark_cases": 3,
|
|
"fusion_gain_gate": [True, False, False],
|
|
"teacher_metadata_without_raw_audio": True,
|
|
"cross_cohort_hidden_404": True,
|
|
"clinical_claim_allowed": False,
|
|
"withdrawal_blocks_voice_processing": True,
|
|
"raw_audio_deleted": True,
|
|
"derived_voice_features_deleted": True,
|
|
"transcript_retained": True,
|
|
},
|
|
}
|
|
|
|
|
|
def parse_args() -> argparse.Namespace:
|
|
parser = argparse.ArgumentParser()
|
|
parser.add_argument("--api-base-url", default="http://127.0.0.1:8014")
|
|
parser.add_argument("--internal-token", required=True)
|
|
parser.add_argument(
|
|
"--benchmark-path",
|
|
default="apps/api/app/data/multimodal_alliance_benchmark_g7.v1.json",
|
|
)
|
|
parser.add_argument(
|
|
"--output",
|
|
default="docs/ops/evidence/multimodal-alliance-live-api-2026-08-06.json",
|
|
)
|
|
parser.add_argument("--request-timeout", type=float, default=30.0)
|
|
return parser.parse_args()
|
|
|
|
|
|
def main() -> None:
|
|
args = parse_args()
|
|
evidence = run(args)
|
|
output = Path(args.output)
|
|
output.parent.mkdir(parents=True, exist_ok=True)
|
|
output.write_text(
|
|
json.dumps(evidence, ensure_ascii=False, indent=2) + "\n", encoding="utf-8"
|
|
)
|
|
print(json.dumps(evidence, ensure_ascii=False, indent=2))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|