Files
cover-letter-haness/skills/draft-korean-it-cover-letter/scripts/harness.py
T

3894 lines
150 KiB
Python
Executable File

#!/usr/bin/env python3
"""Local, dependency-free harness for evidence-grounded Korean cover letters.
The command deliberately keeps the candidate's private workspace in a plain
``.cover-letter`` directory. It never overwrites initialized JSON files, and
the context command only emits material that has passed the explicit consent
gates represented in those files.
"""
from __future__ import annotations
import argparse
import copy
import hashlib
import ipaddress
import json
import os
import re
import shlex
import stat
import sys
import tempfile
import unicodedata
from dataclasses import dataclass
from pathlib import Path
from typing import Any, Iterable, Mapping, Sequence
from urllib.parse import unquote, urlsplit
SCHEMA_VERSION = 1
DATA_DIRECTORY = ".cover-letter"
DOCUMENT_NAMES = ("profile.json", "stories.json", "applications.json", "session.json")
SKILL_DIRECTORY = Path(__file__).resolve().parents[1]
ASSET_DIRECTORY = SKILL_DIRECTORY / "assets"
SEVERITY = {"PASS": 0, "WARN": 1, "BLOCK": 2}
VALID_EVIDENCE_ID = re.compile(r"^[AEFRV]\d{3,}$")
ENTITY_ID_PATTERNS = {
prefix: re.compile(rf"^{prefix}\d{{3,}}$") for prefix in "AEFQRVW"
}
STORY_STATUSES = {"captured", "confirmed", "conflicted", "locked"}
VALUE_STATUSES = {"captured", "confirmed", "locked"}
QUESTION_STATUSES = {
"captured",
"mapped",
"outline_proposed",
"outline_approved",
"drafted",
"verified",
"user_approved",
"needs_recheck",
}
SESSION_STAGES = {
"NEW",
"SCOPED",
"COLLECTING",
"PERSONAL_MODEL_CONFIRMED",
"TARGET_READY",
"QUESTION_READY",
"OUTLINE_APPROVED",
"DRAFTED",
"VERIFIED",
"USER_APPROVED",
"NEEDS_RECHECK",
}
REQUIRED_VERIFICATION_FLAGS = (
"facts_checked",
"ownership_checked",
"privacy_checked",
"question_fit_checked",
"voice_confirmed",
"interview_explainable",
)
VALID_COMMENT = re.compile(
r"<!--\s*evidence\s*:\s*(?P<ids>.*?)\s*-->", re.IGNORECASE | re.DOTALL
)
STANDALONE_VALID_COMMENT_LINE = re.compile(
r"(?im)^[ \t]*<!--\s*evidence\s*:\s*.*?\s*-->[ \t]*(?:\r?\n|$)"
)
EVIDENCE_LIKE_COMMENT = re.compile(
r"<!--\s*evidence\b.*?-->", re.IGNORECASE | re.DOTALL
)
ANY_HTML_COMMENT = re.compile(r"<!--.*?-->", re.DOTALL)
NUMBER_PATTERN = re.compile(r"(?<![A-Za-z0-9])\d+(?:[.,]\d+)?")
NUMBER_CLAIM_PATTERN = re.compile(
r"(?<![A-Za-z0-9])(?P<number>\d+(?:[.,]\d+)?)\s*"
r"(?P<unit>(?:천|만|억|조)\s*(?:명|개|건|회|배|원|사용자|요청)?|"
r"milliseconds?|퍼센트|개월|시간|페이지|사용자|요청|"
r"MiB|GiB|KiB|TiB|TPS|RPS|QPS|%p|ms|[µμ]s|ns|MB|GB|KB|TB|"
r"억원|만원|바이트|개|명|회|건|초|분|일|주|달|월|년|배|줄|원|%|s)?"
r"(?![A-Za-z0-9])",
re.IGNORECASE,
)
KOREAN_NUMBER_CLAIM_PATTERN = re.compile(
r"(?<![가-힣])(?!여러분)(?P<number>(?:"
r"수(?:십|백|천)(?:만|억)?|수(?:만|억)|"
r"한두|두세|서너|네댓|대여섯|예닐곱|일고여덟|"
r"(?:열|스물|스무|서른|마흔|쉰|예순|일흔|여든|아흔)"
r"(?:한|두|세|네|다섯|여섯|일곱|여덟|아홉)?|"
r"[일이삼사오육칠팔구십백천만억조]{2,}|"
r"한|두|세|네|다섯|여섯|일곱|여덟|아홉|반|몇|여러|다수|상당수|수많은)"
r"(?:여)?)(?:의)?\s*"
r"(?P<unit>명|개|건|회|번|차례|배|원|개월|시간|페이지|사용자|요청|"
r"년|달|주|일|분|초|줄|바이트|퍼센트|MB|GB|KB|TB|TPS|RPS|QPS|ms)"
r"(?:으로|에서|에게|부터|까지|와|과|의|를|을|이|가|은|는|도|만)?"
r"(?![가-힣A-Za-z])",
re.IGNORECASE,
)
SINGLE_HANJA_NUMBER_CLAIM_PATTERN = re.compile(
r"(?<![가-힣])(?P<number>[일이삼사오육칠팔구십백천만억조])\s+"
r"(?P<unit>명|개|건|회|번|차례|배|원|개월|시간|페이지|사용자|요청|"
r"년|달|주|일|분|초|줄|바이트|퍼센트|MB|GB|KB|TB|TPS|RPS|QPS|ms)"
r"(?:으로|에서|에게|부터|까지|와|과|의|를|을|이|가|은|는|도|만)?"
r"(?![가-힣A-Za-z])",
re.IGNORECASE,
)
ATTACHED_SINGLE_HANJA_NUMBER_CLAIM_PATTERN = re.compile(
r"(?<![가-힣])(?P<number>[일이삼사오육칠팔구십백천만억조])"
r"(?P<unit>개월|시간|페이지|사용자|요청|바이트|퍼센트)"
r"(?:으로|에서|에게|부터|까지|와|과|의|를|을|이|가|은|는|도|만)?"
r"(?![가-힣])"
)
KOREAN_NUMBER_CLAIM_PATTERNS = (
KOREAN_NUMBER_CLAIM_PATTERN,
SINGLE_HANJA_NUMBER_CLAIM_PATTERN,
ATTACHED_SINGLE_HANJA_NUMBER_CLAIM_PATTERN,
)
TIME_NUMBER_UNITS = {"년", "개월", "달", "월", "주", "일"}
FACT_PATTERN = re.compile(
r"(?:개발|구현|설계|구축|운영|도입|개선|달성|수행|참여|담당|해결|분석|"
r"배포|리팩터링|최적화|수상|근무|리드|협업|제작|완성|선정|통과|"
r"사용|적용|추가|변경|테스트|맡았|만들었|줄였|높였|낮췄|증가시켰|"
r"감소시켰|기여했|경험했)"
)
PLACEHOLDER_PATTERN = re.compile(
r"(?i)(?:\[(?:확인\s*필요|미확인|작성\s*중|추가\s*필요|회사명?|고객명?|프로젝트명?)\]"
r"|\b(?:TODO|TBD|FIXME)\b|\{\{[^{}]+\}\}|\$\{[^{}]+\})"
)
KOREAN_TERM_PATTERN = re.compile(r"[가-힣]{2,}")
KOREAN_PARTICLES = (
"으로부터",
"에게서",
"에서는",
"으로는",
"까지는",
"부터는",
"에게",
"에서",
"으로",
"처럼",
"보다",
"까지",
"부터",
"와",
"과",
"을",
"를",
"은",
"는",
"이",
"가",
"의",
"도",
"만",
"로",
"에",
)
GENERIC_KOREAN_STEMS = (
"저는",
"제가",
"본인",
"직접",
"당시",
"이를",
"통해",
"경험",
"프로젝트",
"과정",
"문제",
"원인",
"결과",
"작업",
"역할",
"확인",
"해결",
"개발",
"구현",
"진행",
"개선",
"추가",
"변경",
"분석",
"수행",
"담당",
"참여",
"사용",
"적용",
"바탕",
"같",
"동일",
"다시",
"현상",
"않",
"먼저",
"이후",
"위해",
"막",
"방지",
"예방",
)
KOREAN_TERM_ALIASES = {
"시연": ("시연", "데모"),
"정지": ("정지", "멈춤", "중단"),
"재현": ("재현", "나타나", "되풀이"),
"재시도": ("재시도", "리트라이"),
}
TEAM_SUBJECT_PATTERN = re.compile(
r"(?:우리\s*팀|팀\s*전체|팀이|팀은|팀의|팀에서는|공동으로|"
r"프로젝트가|프로젝트는|프로젝트의|서비스가|서비스는|서비스의|"
r"시스템이|시스템은|시스템의|작품이)"
)
PERSONAL_SUBJECT_PATTERN = re.compile(
r"(?:제가|저는|나는|내가|저의|내\s+성과|단독|혼자|전담)"
)
INTENTION_PATTERN = re.compile(r"(?:싶습니다|하고자|계획입니다|목표입니다|지향합니다|예정입니다)")
COLLECTIVE_RESULT_MARKERS = (
"수상",
"선정",
"입상",
"우승",
"출시",
"매출",
"계약",
"투자",
"합격",
)
MEASURED_RESULT_MARKERS = (
"개선",
"감소",
"증가",
"단축",
"향상",
"줄였",
"높였",
"낮췄",
"달성",
)
METRIC_RESULT_PATTERN = re.compile(
r"(?:응답\s*시간|처리량|성공률|실패율|오류율|장애율|재발률|매출|"
r"사용자\s*수|요청\s*수|전환율|정확도|지연\s*시간)"
)
RESULT_CLAIM_PATTERN = re.compile(
r"(?:해결|개선|감소|증가|단축|향상|줄였|높였|낮췄|달성|막았|막았습니다|"
r"방지|예방|없앴|사라졌|재현되지|나타나지|재발하지|수상|선정|입상|"
r"우승|출시|매출|계약|투자|합격|성공)"
)
CLICHES = (
"귀사",
"귀사에 기여",
"무한한 가능성",
"끊임없이 노력",
"끊임없이 성장",
"열정과 도전",
"최선을 다하겠습니다",
"성장하는 인재",
"함께 성장",
"기여하겠습니다",
"혁신적인 인재",
"역량을 함양했습니다",
"성장할 수 있었습니다",
"이를 통해",
"이 경험을 통해",
"나아가",
"소통과 협업",
"빠르게 변화하는 시대",
"책임감을 가지고",
"도전 정신",
)
ENDING_PATTERN = re.compile(
r"(하였습니다|했습니다|되었습니다|있었습니다|없었습니다|였습니다|"
r"하겠습니다|되겠습니다|있습니다|없습니다|됩니다|합니다|입니다|"
r"배웠습니다|느꼈습니다|생각합니다|싶습니다)\s*[.!?。!?]?$"
)
SENSITIVE_PATTERNS = {
"주민등록번호로 보이는 값": re.compile(r"(?<!\d)\d{6}\s*-\s*[1-8]\d{6}(?!\d)"),
"휴대전화 번호": re.compile(r"(?<!\d)01[016789][ -]?\d{3,4}[ -]?\d{4}(?!\d)"),
"이메일 주소": re.compile(r"(?i)(?<![\w.+-])[\w.+-]+@[a-z0-9-]+(?:\.[a-z0-9-]+)+"),
"API 키나 접근 토큰으로 보이는 값": re.compile(
r"(?i)(?:sk-[a-z0-9_-]{16,}|gh[pousr]_[a-z0-9]{20,}|AKIA[A-Z0-9]{16}|"
r"(?:api[_ -]?key|access[_ -]?token|secret)\s*[:=]\s*[a-z0-9_./+=-]{12,})"
),
}
URL_PATTERN = re.compile(
r"(?i)\b[a-z][a-z0-9+.-]*(?::[a-z][a-z0-9+.-]*)?://[^\s<>()]+"
)
BARE_ENDPOINT_PATTERN = re.compile(
r"(?i)(?<![A-Za-z0-9_.-])(?:"
r"\[(?:[0-9a-f]{0,4}:){2,}[0-9a-f:.]*\](?::\d{2,5})?|"
r"(?:[0-9a-f]{0,4}:){2,}[0-9a-f]{0,4}|"
r"(?:localhost|(?:\d{1,3}\.){3}\d{1,3}|"
r"[A-Za-z0-9-]+(?:\.[A-Za-z0-9-]+)*\.(?:internal|local|lan))(?::\d{2,5})?|"
r"[A-Za-z][A-Za-z0-9-]{0,62}:\d{2,5}"
r")(?:/[^\s<>()]*)?"
)
SENSITIVE_URL_KEYS = {
"access_token",
"api_key",
"apikey",
"auth",
"authorization",
"code",
"credential",
"hmac",
"id_token",
"jwt",
"key",
"password",
"passwd",
"refresh_token",
"samlresponse",
"secret",
"session",
"session_id",
"sig",
"signature",
"ticket",
"token",
"x-amz-credential",
"x-amz-security-token",
"x-amz-signature",
}
LATIN_TERM_PATTERN = re.compile(
r"(?<![A-Za-z0-9])(?:[A-Za-z][A-Za-z0-9.+#/-]{1,})(?![A-Za-z0-9])"
)
LATIN_TERM_STOPWORDS = {
"a",
"an",
"and",
"as",
"at",
"by",
"for",
"from",
"in",
"into",
"of",
"on",
"or",
"the",
"to",
"with",
}
TECHNOLOGY_ALIASES = {
"Kubernetes": ("kubernetes", "k8s", "쿠버네티스"),
"Docker": ("docker", "도커"),
"Redis": ("redis", "레디스"),
"Kafka": ("kafka", "카프카"),
"Spring": ("spring", "스프링"),
"React": ("react", "리액트"),
"Vue": ("vue", "뷰.js", "vue.js"),
"Terraform": ("terraform", "테라폼"),
"Jenkins": ("jenkins", "젠킨스"),
"Kotlin": ("kotlin", "코틀린"),
"PostgreSQL": ("postgresql", "postgres", "포스트그레스"),
"MySQL": ("mysql", "마이에스큐엘"),
"MongoDB": ("mongodb", "몽고db", "몽고디비"),
"AWS": ("aws", "아마존 웹 서비스"),
"GCP": ("gcp", "구글 클라우드"),
"Azure": ("azure", "애저"),
"LLM": ("llm", "대규모 언어 모델"),
"머신러닝": ("machine learning", "머신러닝"),
"딥러닝": ("deep learning", "딥러닝"),
}
CLAIM_GROUPS = {
"주도·총괄": {
"claim": ("주도", "총괄", "리드", "책임졌"),
"evidence": ("주도", "총괄", "리드", "책임", "직접 결정"),
},
"설계": {
"claim": ("설계", "아키텍처", "모델링"),
"evidence": ("설계", "아키텍처", "모델링", "구조를 정"),
},
"구현·개발": {
"claim": ("구현", "개발", "제작", "만들었"),
"evidence": ("구현", "개발", "제작", "만들", "작성"),
},
"배포·운영·출시": {
"claim": ("배포", "운영", "출시"),
"evidence": ("배포", "운영", "출시", "프로덕션"),
},
"기술 사용·도입": {
"claim": ("도입", "사용", "적용"),
"evidence": ("도입", "사용", "적용", "채택"),
},
"개선·최적화": {
"claim": ("개선", "최적화", "단축", "줄였", "높였", "낮췄", "향상"),
"evidence": (
"개선",
"최적화",
"단축",
"줄였",
"줄임",
"감소",
"높였",
"낮췄",
"향상",
"빨라",
),
},
"문제 해결": {
"claim": ("해결", "수정", "고쳤"),
"evidence": (
"해결",
"수정",
"고쳤",
"재현되지",
"오류가 사라",
"변경",
"조정",
"원인을 좁",
),
},
"수상·선정": {
"claim": ("수상", "선정", "입상", "합격"),
"evidence": ("수상", "선정", "입상", "합격"),
},
"단독·전담": {
"claim": ("단독", "혼자", "전담", "전부 맡", "모두 맡"),
"evidence": ("단독", "혼자", "전담", "전부", "모두", "개인 프로젝트"),
},
}
POSTING_CONTEXT_FIELDS = ("text", "source", "captured_at", "deadline")
STORY_OWNERSHIP_FIELDS = (
"user_role",
"decisions",
"actions",
"alternatives",
"reflection",
"skills",
)
class HarnessError(Exception):
"""An expected, user-actionable harness error."""
@dataclass(frozen=True)
class Sentence:
text: str
start: int
end: int
line: int
evidence_ids: tuple[str, ...]
def _json_print(value: Any) -> None:
print(json.dumps(value, ensure_ascii=False, indent=2))
def _json_digest(value: Any) -> str:
payload = json.dumps(
value,
ensure_ascii=False,
sort_keys=True,
separators=(",", ":"),
).encode("utf-8")
return hashlib.sha256(payload).hexdigest()
def _resolve_path(value: str | Path, label: str = "경로") -> Path:
try:
return Path(value).expanduser().resolve()
except (OSError, RuntimeError) as exc:
raise HarnessError(f"{label}를 확인할 수 없습니다: {exc}") from exc
def _data_directory(workspace: str | Path | None) -> Path:
root = _resolve_path(workspace or ".", "워크스페이스 경로")
return root if root.name == DATA_DIRECTORY else root / DATA_DIRECTORY
def _discover_data_directory(start: Path) -> Path | None:
candidate = _resolve_path(start, "초안 경로")
if candidate.is_file():
candidate = candidate.parent
for parent in (candidate, *candidate.parents):
if parent.name == DATA_DIRECTORY and parent.is_dir():
return parent
nested = parent / DATA_DIRECTORY
if nested.is_dir():
return nested
return None
def _read_json(path: Path) -> Any:
try:
with path.open("r", encoding="utf-8") as handle:
return json.load(handle)
except FileNotFoundError as exc:
raise HarnessError(f"필수 파일이 없습니다: {path}") from exc
except json.JSONDecodeError as exc:
raise HarnessError(
f"JSON 형식이 올바르지 않습니다: {path} ({exc.lineno}{exc.colno}열)"
) from exc
except UnicodeError as exc:
raise HarnessError(f"JSON 파일은 UTF-8이어야 합니다: {path}") from exc
except OSError as exc:
raise HarnessError(f"파일을 읽을 수 없습니다: {path} ({exc})") from exc
def _load_workspace(data_dir: Path) -> dict[str, Any]:
return {name: _read_json(data_dir / name) for name in DOCUMENT_NAMES}
def _write_new_file(source: Path, destination: Path) -> bool:
"""Copy an asset with exclusive creation; return False when it exists."""
try:
payload = source.read_bytes()
descriptor = os.open(destination, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
with os.fdopen(descriptor, "wb") as handle:
handle.write(payload)
return True
except FileExistsError:
return False
except OSError as exc:
raise HarnessError(f"초기 파일을 만들 수 없습니다: {destination} ({exc})") from exc
def _write_private_new(destination: Path, payload: bytes) -> bool:
try:
descriptor = os.open(destination, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
with os.fdopen(descriptor, "wb") as handle:
handle.write(payload)
return True
except FileExistsError:
return False
except OSError as exc:
raise HarnessError(f"보호 파일을 만들 수 없습니다: {destination} ({exc})") from exc
def _write_private_atomic_bytes(destination: Path, payload: bytes) -> None:
descriptor, temporary_name = tempfile.mkstemp(
prefix=f".{destination.name}.", dir=destination.parent
)
temporary_path = Path(temporary_name)
try:
os.fchmod(descriptor, 0o600)
with os.fdopen(descriptor, "wb") as handle:
handle.write(payload)
handle.flush()
os.fsync(handle.fileno())
os.replace(temporary_path, destination)
except OSError as exc:
raise HarnessError(f"보호 파일을 안전하게 저장할 수 없습니다: {destination} ({exc})") from exc
finally:
if temporary_path.exists():
temporary_path.unlink()
def _set_private_mode(path: Path, mode: int, label: str) -> None:
try:
if path.is_symlink():
raise HarnessError(f"{label}이 심볼릭 링크여서 사용할 수 없습니다: {path}")
path.chmod(mode)
except OSError as exc:
raise HarnessError(f"{label} 권한을 보호할 수 없습니다: {path} ({exc})") from exc
def _write_json_private_atomic(destination: Path, value: Any) -> None:
"""Replace one workspace JSON file without exposing a partial write."""
destination.parent.mkdir(parents=True, mode=0o700, exist_ok=True)
descriptor, temporary_name = tempfile.mkstemp(
prefix=f".{destination.name}.", dir=destination.parent
)
temporary_path = Path(temporary_name)
try:
os.fchmod(descriptor, 0o600)
with os.fdopen(descriptor, "w", encoding="utf-8") as handle:
json.dump(value, handle, ensure_ascii=False, indent=2)
handle.write("\n")
handle.flush()
os.fsync(handle.fileno())
os.replace(temporary_path, destination)
except OSError as exc:
raise HarnessError(f"JSON 파일을 안전하게 저장할 수 없습니다: {destination} ({exc})") from exc
finally:
if temporary_path.exists():
temporary_path.unlink()
def initialize_workspace(workspace: str | Path | None = None) -> dict[str, Any]:
data_dir = _data_directory(workspace)
try:
data_dir.mkdir(parents=True, mode=0o700, exist_ok=True)
except OSError as exc:
raise HarnessError(f"워크스페이스를 만들 수 없습니다: {data_dir} ({exc})") from exc
_set_private_mode(data_dir, 0o700, "워크스페이스")
created: list[str] = []
skipped: list[str] = []
for name in DOCUMENT_NAMES:
if _write_new_file(ASSET_DIRECTORY / name, data_dir / name):
created.append(name)
else:
skipped.append(name)
_set_private_mode(data_dir / name, 0o600, name)
for name in ("drafts", "exports"):
path = data_dir / name
existed = path.exists()
path.mkdir(parents=True, mode=0o700, exist_ok=True)
_set_private_mode(path, 0o700, f"{name} 디렉터리")
(skipped if existed else created).append(f"{name}/")
privacy_ignore = data_dir / ".gitignore"
if _write_private_new(privacy_ignore, b"*\n"):
created.append(".gitignore")
else:
skipped.append(".gitignore")
try:
ignore_payload = privacy_ignore.read_bytes()
except OSError as exc:
raise HarnessError(f".gitignore를 읽을 수 없습니다: {privacy_ignore} ({exc})") from exc
if not any(line.strip() == b"*" for line in ignore_payload.splitlines()):
separator = b"" if not ignore_payload or ignore_payload.endswith(b"\n") else b"\n"
_write_private_atomic_bytes(privacy_ignore, ignore_payload + separator + b"*\n")
_set_private_mode(privacy_ignore, 0o600, ".gitignore")
return {
"status": "PASS",
"workspace": str(data_dir),
"created": created,
"skipped": skipped,
}
def _is_object(value: Any) -> bool:
return isinstance(value, dict)
def _is_list(value: Any) -> bool:
return isinstance(value, list)
def _has_text(value: Any) -> bool:
return isinstance(value, str) and bool(value.strip())
def _has_text_items(value: Any) -> bool:
return _is_list(value) and bool(value) and all(_has_text(item) for item in value)
def _is_text_list(value: Any) -> bool:
return _is_list(value) and all(_has_text(item) for item in value)
def _is_enum(value: Any, allowed: set[str]) -> bool:
return isinstance(value, str) and value in allowed
def _safe_reference_label(value: Any) -> str:
if isinstance(value, str) and re.fullmatch(r"[AEFQRVW]\d{3,}", value):
return value
return "[형식이 잘못된 값]"
def _unsafe_unicode_locations(value: Any, location: str) -> list[str]:
locations: list[str] = []
if isinstance(value, str):
if value != unicodedata.normalize("NFC", value) or any(
unicodedata.category(character) == "Cf"
or (
unicodedata.category(character) == "Cc"
and character not in "\n\r\t"
)
for character in value
):
locations.append(location)
elif isinstance(value, Mapping):
for index, (key, item) in enumerate(value.items()):
key_location = f"{location}.<key:{index}>"
locations.extend(_unsafe_unicode_locations(key, key_location))
locations.extend(_unsafe_unicode_locations(item, f"{location}.<value:{index}>"))
elif isinstance(value, list):
for index, item in enumerate(value):
locations.extend(_unsafe_unicode_locations(item, f"{location}[{index}]"))
return locations
def _add_issue(
issues: list[dict[str, str]], status: str, code: str, message: str
) -> None:
issues.append({"status": status, "code": code, "message": message})
def validate_workspace(data_dir: Path) -> dict[str, Any]:
issues: list[dict[str, str]] = []
permission_targets = {
data_dir: 0o700,
data_dir / "drafts": 0o700,
data_dir / "exports": 0o700,
**{data_dir / name: 0o600 for name in DOCUMENT_NAMES},
data_dir / ".gitignore": 0o600,
}
for path, expected_mode in permission_targets.items():
try:
if path.is_symlink():
_add_issue(
issues,
"BLOCK",
"PRIVACY_SYMLINK",
f"개인 작업 경로가 심볼릭 링크이면 안 됩니다: {path.name}",
)
elif path.exists() and stat.S_IMODE(path.stat().st_mode) != expected_mode:
_add_issue(
issues,
"BLOCK",
"PRIVACY_MODE",
f"{path.name} 권한은 {oct(expected_mode)}이어야 합니다. init으로 복구하세요.",
)
except OSError as exc:
_add_issue(
issues,
"BLOCK",
"PRIVACY_MODE",
f"{path.name} 권한을 확인할 수 없습니다: {exc}",
)
privacy_ignore = data_dir / ".gitignore"
try:
ignore_payload = privacy_ignore.read_bytes()
if not any(line.strip() == b"*" for line in ignore_payload.splitlines()):
_add_issue(
issues,
"BLOCK",
"PRIVACY_IGNORE",
".cover-letter/.gitignore에 전체 제외 규칙 *가 필요합니다. init으로 복구하세요.",
)
except FileNotFoundError:
_add_issue(
issues,
"BLOCK",
"PRIVACY_IGNORE",
".cover-letter/.gitignore가 필요합니다. init으로 복구하세요.",
)
except OSError as exc:
_add_issue(
issues,
"BLOCK",
"PRIVACY_IGNORE",
f".gitignore를 확인할 수 없습니다: {exc}",
)
documents: dict[str, Any] = {}
for name in DOCUMENT_NAMES:
path = data_dir / name
try:
document = _read_json(path)
except HarnessError as exc:
_add_issue(issues, "BLOCK", "DOCUMENT_READ", str(exc))
continue
documents[name] = document
unsafe_locations = _unsafe_unicode_locations(document, name)
if unsafe_locations:
_add_issue(
issues,
"BLOCK",
"UNSAFE_WORKSPACE_UNICODE",
f"{name}에 정규화되지 않은 문자나 보이지 않는 제어 문자가 있습니다: "
+ ", ".join(unsafe_locations[:5]),
)
if not _is_object(document):
_add_issue(issues, "BLOCK", "DOCUMENT_TYPE", f"{name} 최상위 값은 객체여야 합니다.")
continue
if document.get("schema_version") != SCHEMA_VERSION:
_add_issue(
issues,
"BLOCK",
"SCHEMA_VERSION",
f"{name}.schema_version은 {SCHEMA_VERSION}이어야 합니다.",
)
profile = documents.get("profile.json")
stories_doc = documents.get("stories.json")
applications_doc = documents.get("applications.json")
session = documents.get("session.json")
motivation_references: list[str] = []
if _is_object(profile):
for field, expected in (
("candidate", dict),
("values", list),
("voice", dict),
("privacy", dict),
("confirmed", bool),
):
if not isinstance(profile.get(field), expected):
_add_issue(
issues,
"BLOCK",
"PROFILE_FIELD",
f"profile.json의 {field} 필드 형식이 올바르지 않습니다.",
)
candidate = profile.get("candidate", {})
if _is_object(candidate):
if not _is_list(candidate.get("target_roles")):
_add_issue(
issues,
"BLOCK",
"PROFILE_FIELD",
"profile.json.candidate.target_roles는 배열이어야 합니다.",
)
if not _is_object(candidate.get("motivation")):
_add_issue(
issues,
"BLOCK",
"PROFILE_FIELD",
"profile.json.candidate.motivation은 객체여야 합니다.",
)
voice = profile.get("voice", {})
if _is_object(voice):
if not _is_list(voice.get("samples")):
_add_issue(
issues,
"BLOCK",
"PROFILE_FIELD",
"profile.json.voice.samples는 배열이어야 합니다.",
)
for field in ("preferred_tone", "avoid_phrases"):
if not _is_text_list(voice.get(field)):
_add_issue(
issues,
"BLOCK",
"PROFILE_FIELD",
f"profile.json.voice.{field}는 문자열 배열이어야 합니다.",
)
privacy = profile.get("privacy", {})
if _is_object(privacy):
for field in ("do_not_use", "redactions"):
if not _is_text_list(privacy.get(field)):
_add_issue(
issues,
"BLOCK",
"PROFILE_FIELD",
f"profile.json.privacy.{field}는 문자열 배열이어야 합니다.",
)
if profile.get("confirmed") is True and _is_object(candidate):
if not _has_text(candidate.get("career_level")):
_add_issue(
issues,
"BLOCK",
"PROFILE_COMPLETENESS",
"확정 profile에는 candidate.career_level이 필요합니다.",
)
if not _has_text_items(candidate.get("target_roles")):
_add_issue(
issues,
"BLOCK",
"PROFILE_COMPLETENESS",
"확정 profile에는 하나 이상의 candidate.target_roles가 필요합니다.",
)
if not _has_text(candidate.get("current_status")):
_add_issue(
issues,
"BLOCK",
"PROFILE_COMPLETENESS",
"확정 profile에는 candidate.current_status가 필요합니다.",
)
motivation = candidate.get("motivation", {})
if not _is_object(motivation) or any(
not _has_text(motivation.get(field)) for field in ("why_it", "why_role")
):
_add_issue(
issues,
"BLOCK",
"PROFILE_COMPLETENESS",
"확정 profile에는 확인된 why_it과 why_role이 필요합니다.",
)
raw_motivation_refs = (
motivation.get("evidence_ids") if _is_object(motivation) else None
)
if not _has_text_items(raw_motivation_refs):
_add_issue(
issues,
"BLOCK",
"MOTIVATION_EVIDENCE",
"확정 motivation에는 하나 이상의 E/V evidence_ids가 필요합니다.",
)
else:
for ref_index, ref in enumerate(raw_motivation_refs):
if re.fullmatch(r"[EV]\d{3,}", ref) is None:
_add_issue(
issues,
"BLOCK",
"MOTIVATION_EVIDENCE",
f"motivation.evidence_ids[{ref_index}]는 E 또는 V ID여야 합니다.",
)
else:
motivation_references.append(ref)
stories = stories_doc.get("stories") if _is_object(stories_doc) else None
if not _is_list(stories):
if stories_doc is not None:
_add_issue(issues, "BLOCK", "STORIES_FIELD", "stories.json.stories는 배열이어야 합니다.")
stories = []
applications = (
applications_doc.get("applications") if _is_object(applications_doc) else None
)
if not _is_list(applications):
if applications_doc is not None:
_add_issue(
issues,
"BLOCK",
"APPLICATIONS_FIELD",
"applications.json.applications는 배열이어야 합니다.",
)
applications = []
if _is_object(session):
if not isinstance(session.get("stage"), str):
_add_issue(
issues,
"BLOCK",
"SESSION_FIELD",
"session.json의 stage 필드는 문자열이어야 합니다.",
)
elif not _is_enum(session.get("stage"), SESSION_STAGES):
_add_issue(
issues,
"BLOCK",
"SESSION_STAGE",
"session.json.stage가 허용된 상태가 아닙니다.",
)
for field in ("active_application_id", "active_question_id"):
if session.get(field) is not None and not isinstance(session.get(field), str):
_add_issue(
issues,
"BLOCK",
"SESSION_FIELD",
f"session.json의 {field} 필드는 문자열이어야 합니다.",
)
if not _is_list(session.get("pending_confirmations")):
_add_issue(
issues,
"BLOCK",
"SESSION_FIELD",
"session.json.pending_confirmations는 배열이어야 합니다.",
)
if session.get("updated_at") is not None and not isinstance(
session.get("updated_at"), str
):
_add_issue(
issues,
"BLOCK",
"SESSION_FIELD",
"session.json.updated_at은 문자열 또는 null이어야 합니다.",
)
identifiers: dict[str, str] = {}
def register(identifier: Any, location: str, prefix: str) -> None:
if not isinstance(identifier, str) or not identifier.strip():
_add_issue(issues, "BLOCK", "MISSING_ID", f"{location}에 id가 필요합니다.")
return
if ENTITY_ID_PATTERNS[prefix].fullmatch(identifier) is None:
_add_issue(
issues,
"BLOCK",
"INVALID_ID",
f"{location}.id는 {prefix}와 세 자리 이상 숫자 형식이어야 합니다.",
)
if identifier in identifiers:
_add_issue(
issues,
"BLOCK",
"DUPLICATE_ID",
f"id {_safe_reference_label(identifier)}{identifiers[identifier]}{location}에서 중복됩니다.",
)
else:
identifiers[identifier] = location
def collect_references(
raw: Any,
location: str,
pattern: re.Pattern[str],
destination: list[tuple[str, str]],
) -> list[str]:
if not _is_list(raw):
_add_issue(issues, "BLOCK", "REFERENCE_TYPE", f"{location}는 배열이어야 합니다.")
return []
valid: list[str] = []
for ref_index, ref in enumerate(raw):
if not isinstance(ref, str) or pattern.fullmatch(ref) is None:
_add_issue(
issues,
"BLOCK",
"INVALID_REFERENCE",
f"{location}[{ref_index}]의 참조 ID 형식이 올바르지 않습니다.",
)
continue
valid.append(ref)
destination.append((ref, location))
return valid
values = profile.get("values", []) if _is_object(profile) else []
value_ids: set[str] = set()
eligible_motivation_ids: set[str] = set()
for index, value in enumerate(values if _is_list(values) else []):
if not _is_object(value):
_add_issue(issues, "BLOCK", "VALUE_TYPE", f"values[{index}]는 객체여야 합니다.")
continue
register(value.get("id"), f"values[{index}]", "V")
if isinstance(value.get("id"), str):
value_ids.add(value["id"])
if (
_is_enum(value.get("status"), {"confirmed", "locked"})
and value.get("use_permission") == "allowed"
):
eligible_motivation_ids.add(value["id"])
if not _is_enum(value.get("status"), VALUE_STATUSES):
_add_issue(
issues,
"BLOCK",
"VALUE_STATUS",
f"values[{index}].status가 올바르지 않습니다.",
)
if not _is_enum(value.get("use_permission"), {"ask", "allowed", "blocked"}):
_add_issue(
issues,
"BLOCK",
"USE_PERMISSION",
f"values[{index}].use_permission이 올바르지 않습니다.",
)
if _is_enum(value.get("status"), {"confirmed", "locked"}):
if any(
not _has_text(value.get(field))
for field in ("statement", "origin", "behavior")
) or not _has_text_items(value.get("story_ids")):
_add_issue(
issues,
"BLOCK",
"VALUE_COMPLETENESS",
f"values[{index}]의 확정 가치에는 statement, origin, behavior와 story_ids가 필요합니다.",
)
voice_samples = (
profile.get("voice", {}).get("samples", [])
if _is_object(profile) and _is_object(profile.get("voice"))
else []
)
for index, sample in enumerate(voice_samples if _is_list(voice_samples) else []):
if not _is_object(sample):
_add_issue(
issues,
"BLOCK",
"VOICE_SAMPLE_TYPE",
f"voice.samples[{index}]는 객체여야 합니다.",
)
continue
register(sample.get("id"), f"voice.samples[{index}]", "W")
if not isinstance(sample.get("user_authored"), bool):
_add_issue(
issues,
"BLOCK",
"VOICE_SAMPLE_FIELD",
f"voice.samples[{index}].user_authored는 불리언이어야 합니다.",
)
if not _is_enum(sample.get("use_permission"), {"ask", "allowed", "blocked"}):
_add_issue(
issues,
"BLOCK",
"USE_PERMISSION",
f"voice.samples[{index}].use_permission이 올바르지 않습니다.",
)
if sample.get("use_permission") == "allowed" and (
not _has_text(sample.get("kind")) or not _has_text(sample.get("text"))
):
_add_issue(
issues,
"BLOCK",
"VOICE_SAMPLE_FIELD",
f"voice.samples[{index}]의 허용 표본에는 kind와 text가 필요합니다.",
)
story_ids: set[str] = set()
evidence_ids: set[str] = set()
story_references: list[tuple[str, str]] = []
value_references: list[tuple[str, str]] = []
outline_references: list[tuple[str, str]] = []
for index, story in enumerate(stories):
if not _is_object(story):
_add_issue(issues, "BLOCK", "STORY_TYPE", f"stories[{index}]는 객체여야 합니다.")
continue
location = f"stories[{index}]"
register(story.get("id"), location, "E")
if isinstance(story.get("id"), str):
story_ids.add(story["id"])
if _eligible_story(story):
eligible_motivation_ids.add(story["id"])
permission = story.get("use_permission", "ask")
if not _is_enum(permission, {"ask", "allowed", "blocked"}):
_add_issue(
issues,
"BLOCK",
"USE_PERMISSION",
f"{location}.use_permission은 ask, allowed, blocked 중 하나여야 합니다.",
)
if not _is_enum(story.get("status"), STORY_STATUSES):
_add_issue(
issues,
"BLOCK",
"STORY_STATUS",
f"{location}.status가 올바르지 않습니다.",
)
for list_field in (
"constraints",
"decisions",
"actions",
"alternatives",
"skills",
"value_ids",
"uncertainties",
"conflicts",
):
if list_field in story and not _is_text_list(story.get(list_field)):
_add_issue(
issues,
"BLOCK",
"STORY_FIELD",
f"{location}.{list_field}는 문자열 배열이어야 합니다.",
)
unresolved = [
field
for field in ("uncertainties", "conflicts")
if _is_list(story.get(field)) and bool(story.get(field))
]
if unresolved and _is_enum(story.get("status"), {"confirmed", "locked"}):
_add_issue(
issues,
"BLOCK",
"UNRESOLVED_STORY",
f"{location}에 미해결 uncertainties/conflicts가 있으면 confirmed/locked일 수 없습니다.",
)
outcomes = story.get("outcomes")
if outcomes is not None and not _is_object(outcomes):
_add_issue(
issues,
"BLOCK",
"OUTCOME_TYPE",
f"{location}.outcomes는 객체여야 합니다.",
)
elif _is_object(outcomes):
for outcome_field in ("measured", "observed"):
if outcome_field in outcomes and not _is_text_list(outcomes.get(outcome_field)):
_add_issue(
issues,
"BLOCK",
"OUTCOME_FIELD",
f"{location}.outcomes.{outcome_field}는 문자열 배열이어야 합니다.",
)
if "attribution" in outcomes and not isinstance(outcomes.get("attribution"), str):
_add_issue(
issues,
"BLOCK",
"OUTCOME_FIELD",
f"{location}.outcomes.attribution은 문자열이어야 합니다.",
)
if "owner_scope" in outcomes and not _is_enum(
outcomes.get("owner_scope"), {"user", "team", "shared"}
):
_add_issue(
issues,
"BLOCK",
"STORY_OWNERSHIP",
f"{location}.outcomes.owner_scope가 올바르지 않습니다.",
)
if (
_is_enum(story.get("status"), {"confirmed", "locked"})
and permission == "allowed"
):
if not _has_text(story.get("title")) or not _has_text(story.get("user_role")):
_add_issue(
issues,
"BLOCK",
"STORY_COMPLETENESS",
f"{location}의 허용된 확정 경험에는 title과 user_role이 필요합니다.",
)
if not _has_text_items(story.get("actions")):
_add_issue(
issues,
"BLOCK",
"STORY_COMPLETENESS",
f"{location}의 허용된 확정 경험에는 하나 이상의 구체 행동이 필요합니다.",
)
outcomes = story.get("outcomes")
outcome_items: list[Any] = []
if _is_object(outcomes):
for outcome_field in ("measured", "observed"):
values_for_field = outcomes.get(outcome_field, [])
if _is_list(values_for_field):
outcome_items.extend(values_for_field)
if not any(_has_text(item) for item in outcome_items):
_add_issue(
issues,
"BLOCK",
"STORY_COMPLETENESS",
f"{location}의 허용된 확정 경험에는 측정 또는 관찰 결과가 필요합니다.",
)
if not _is_object(outcomes) or not _has_text(outcomes.get("attribution")):
_add_issue(
issues,
"BLOCK",
"STORY_OWNERSHIP",
f"{location}.outcomes.attribution에 팀 결과와 본인 기여 범위를 적어야 합니다.",
)
if not _is_object(outcomes) or not _is_enum(
outcomes.get("owner_scope"), {"user", "team", "shared"}
):
_add_issue(
issues,
"BLOCK",
"STORY_OWNERSHIP",
f"{location}.outcomes.owner_scope는 user, team, shared 중 하나여야 합니다.",
)
collect_references(
story.get("value_ids", []),
f"{location}.value_ids",
ENTITY_ID_PATTERNS["V"],
value_references,
)
evidence = story.get("evidence", [])
if not _is_list(evidence):
_add_issue(issues, "BLOCK", "EVIDENCE_TYPE", f"{location}.evidence는 배열이어야 합니다.")
continue
for evidence_index, item in enumerate(evidence):
evidence_location = f"{location}.evidence[{evidence_index}]"
if not _is_object(item):
_add_issue(
issues, "BLOCK", "EVIDENCE_TYPE", f"{evidence_location}는 객체여야 합니다."
)
continue
register(item.get("id"), evidence_location, "F")
if isinstance(item.get("id"), str):
evidence_ids.add(item["id"])
if "verified" in item and not isinstance(item.get("verified"), bool):
_add_issue(
issues,
"BLOCK",
"EVIDENCE_VERIFIED",
f"{evidence_location}.verified는 불리언이어야 합니다.",
)
if "status" in item and not _is_enum(item.get("status"), STORY_STATUSES):
_add_issue(
issues,
"BLOCK",
"EVIDENCE_STATUS",
f"{evidence_location}.status가 올바르지 않습니다.",
)
for text_field in ("claim", "source"):
if text_field in item and not isinstance(item.get(text_field), str):
_add_issue(
issues,
"BLOCK",
"EVIDENCE_FIELD",
f"{evidence_location}.{text_field}는 문자열이어야 합니다.",
)
if "owner_scope" in item and not _is_enum(
item.get("owner_scope"), {"user", "team", "shared"}
):
_add_issue(
issues,
"BLOCK",
"STORY_OWNERSHIP",
f"{evidence_location}.owner_scope가 올바르지 않습니다.",
)
if not _is_enum(
item.get("use_permission", permission), {"ask", "allowed", "blocked"}
):
_add_issue(
issues,
"BLOCK",
"USE_PERMISSION",
f"{evidence_location}.use_permission이 올바르지 않습니다.",
)
if "owner_scope" in item and not _is_enum(
item.get("owner_scope"), {"user", "team", "shared"}
):
_add_issue(
issues,
"BLOCK",
"EVIDENCE_OWNERSHIP",
f"{evidence_location}.owner_scope가 올바르지 않습니다.",
)
for index, value in enumerate(values if _is_list(values) else []):
if not _is_object(value):
continue
collect_references(
value.get("story_ids", []),
f"values[{index}].story_ids",
ENTITY_ID_PATTERNS["E"],
story_references,
)
app_ids: set[str] = set()
question_ids: set[str] = set()
question_parent: dict[str, str] = {}
for app_index, app in enumerate(applications):
if not _is_object(app):
_add_issue(
issues, "BLOCK", "APPLICATION_TYPE", f"applications[{app_index}]는 객체여야 합니다."
)
continue
app_location = f"applications[{app_index}]"
register(app.get("id"), app_location, "A")
if isinstance(app.get("id"), str):
app_ids.add(app["id"])
if not isinstance(app.get("confirmed"), bool):
_add_issue(
issues,
"BLOCK",
"APPLICATION_CONFIRMATION",
f"{app_location}.confirmed는 불리언이어야 합니다.",
)
if app.get("confirmed") is True:
if not _has_text(app.get("company")) or not _has_text(app.get("role")):
_add_issue(
issues,
"BLOCK",
"APPLICATION_COMPLETENESS",
f"{app_location}의 확정 지원처에는 company와 role이 필요합니다.",
)
posting = app.get("posting")
posting_text = posting.get("text") if _is_object(posting) else None
if not _has_text(posting_text) and not app.get("requirements"):
_add_issue(
issues,
"BLOCK",
"APPLICATION_COMPLETENESS",
f"{app_location}의 확정 지원처에는 공고 원문 또는 요구사항이 필요합니다.",
)
if not _is_object(posting) or (
not _has_text(posting.get("source"))
or not _has_text(posting.get("captured_at"))
):
_add_issue(
issues,
"BLOCK",
"APPLICATION_SOURCE",
f"{app_location}의 확정 공고·요구사항에는 posting.source와 captured_at이 필요합니다.",
)
posting = app.get("posting")
if posting is not None and not _is_object(posting):
_add_issue(
issues,
"BLOCK",
"APPLICATION_FIELD",
f"{app_location}.posting은 객체여야 합니다.",
)
elif _is_object(posting):
for posting_field in POSTING_CONTEXT_FIELDS:
if posting_field in posting and not isinstance(posting.get(posting_field), str):
_add_issue(
issues,
"BLOCK",
"APPLICATION_FIELD",
f"{app_location}.posting.{posting_field}는 문자열이어야 합니다.",
)
requirements = app.get("requirements", [])
if not _is_list(requirements):
_add_issue(
issues, "BLOCK", "REQUIREMENTS_TYPE", f"{app_location}.requirements는 배열이어야 합니다."
)
else:
for requirement_index, requirement in enumerate(requirements):
requirement_location = f"{app_location}.requirements[{requirement_index}]"
if not _is_object(requirement):
_add_issue(
issues, "BLOCK", "REQUIREMENT_TYPE", f"{requirement_location}는 객체여야 합니다."
)
continue
register(requirement.get("id"), requirement_location, "R")
if not _has_text(requirement.get("text")):
_add_issue(
issues,
"BLOCK",
"REQUIREMENT_COMPLETENESS",
f"{requirement_location}.text가 필요합니다.",
)
if "priority" in requirement and not isinstance(
requirement.get("priority"), str
):
_add_issue(
issues,
"BLOCK",
"REQUIREMENT_COMPLETENESS",
f"{requirement_location}.priority는 문자열이어야 합니다.",
)
collect_references(
requirement.get("story_ids", []),
f"{requirement_location}.story_ids",
ENTITY_ID_PATTERNS["E"],
story_references,
)
questions = app.get("questions", [])
if not _is_list(questions):
_add_issue(issues, "BLOCK", "QUESTIONS_TYPE", f"{app_location}.questions는 배열이어야 합니다.")
continue
for question_index, question in enumerate(questions):
question_location = f"{app_location}.questions[{question_index}]"
if not _is_object(question):
_add_issue(issues, "BLOCK", "QUESTION_TYPE", f"{question_location}는 객체여야 합니다.")
continue
register(question.get("id"), question_location, "Q")
if isinstance(question.get("id"), str):
question_ids.add(question["id"])
if isinstance(app.get("id"), str):
question_parent[question["id"]] = app["id"]
refs = collect_references(
question.get("story_ids", []),
f"{question_location}.story_ids",
ENTITY_ID_PATTERNS["E"],
story_references,
)
limit = question.get("character_limit")
if limit is not None and (not isinstance(limit, int) or isinstance(limit, bool) or limit <= 0):
_add_issue(
issues,
"BLOCK",
"CHARACTER_LIMIT",
f"{question_location}.character_limit은 양의 정수 또는 null이어야 합니다.",
)
minimum = question.get("character_minimum")
if minimum is not None and (
not isinstance(minimum, int) or isinstance(minimum, bool) or minimum < 0
):
_add_issue(
issues,
"BLOCK",
"CHARACTER_MINIMUM",
f"{question_location}.character_minimum은 0 이상의 정수 또는 null이어야 합니다.",
)
if (
isinstance(minimum, int)
and not isinstance(minimum, bool)
and isinstance(limit, int)
and not isinstance(limit, bool)
and minimum > limit
):
_add_issue(
issues,
"BLOCK",
"CHARACTER_RANGE",
f"{question_location}의 최소 글자 수가 상한보다 큽니다.",
)
required_format_value = question.get("required_format")
if required_format_value is not None and not _is_enum(
required_format_value, {"plain_text", "markdown"}
):
_add_issue(
issues,
"BLOCK",
"REQUIRED_FORMAT",
f"{question_location}.required_format은 plain_text, markdown 또는 null이어야 합니다.",
)
if not isinstance(question.get("outline_approved", False), bool):
_add_issue(
issues, "BLOCK", "OUTLINE_APPROVAL", f"{question_location}.outline_approved는 불리언이어야 합니다."
)
if not isinstance(question.get("count_spaces", True), bool):
_add_issue(
issues,
"BLOCK",
"COUNT_SPACES",
f"{question_location}.count_spaces는 불리언이어야 합니다.",
)
if not _is_enum(question.get("status"), QUESTION_STATUSES):
_add_issue(
issues,
"BLOCK",
"QUESTION_STATUS",
f"{question_location}.status가 올바르지 않습니다.",
)
if not isinstance(question.get("prompt"), str):
_add_issue(
issues,
"BLOCK",
"QUESTION_COMPLETENESS",
f"{question_location}.prompt는 문자열이어야 합니다.",
)
elif question.get("status") != "captured" and not _has_text(question.get("prompt")):
_add_issue(
issues,
"BLOCK",
"QUESTION_COMPLETENESS",
f"{question_location}의 진행된 문항에는 prompt가 필요합니다.",
)
if question.get("outline_approved") is True and not _is_enum(
question.get("status"),
{
"outline_approved",
"drafted",
"verified",
"user_approved",
"needs_recheck",
},
):
_add_issue(
issues,
"BLOCK",
"QUESTION_STATE_CONFLICT",
f"{question_location}의 승인 개요와 문항 상태가 서로 맞지 않습니다.",
)
outline = question.get("outline")
if not _is_object(outline):
_add_issue(
issues,
"BLOCK",
"OUTLINE_TYPE",
f"{question_location}.outline은 객체여야 합니다.",
)
else:
if not isinstance(outline.get("thesis"), str):
_add_issue(
issues,
"BLOCK",
"OUTLINE_FIELD",
f"{question_location}.outline.thesis는 문자열이어야 합니다.",
)
if not _is_text_list(outline.get("beats")):
_add_issue(
issues,
"BLOCK",
"OUTLINE_FIELD",
f"{question_location}.outline.beats는 문자열 배열이어야 합니다.",
)
approved_evidence = collect_references(
outline.get("evidence_ids"),
f"{question_location}.outline.evidence_ids",
VALID_EVIDENCE_ID,
outline_references,
)
if _is_list(outline.get("evidence_ids")):
if question.get("outline_approved") is True:
selected_story_ids = set(refs)
if (
not _has_text(outline.get("thesis"))
or not _has_text_items(outline.get("beats"))
or not selected_story_ids
or not approved_evidence
):
_add_issue(
issues,
"BLOCK",
"OUTLINE_COMPLETENESS",
f"{question_location}의 승인 개요에는 핵심 답, 전개, 경험, 근거가 필요합니다.",
)
if not selected_story_ids <= set(approved_evidence):
_add_issue(
issues,
"BLOCK",
"OUTLINE_EVIDENCE",
f"{question_location}의 선택 경험이 outline.evidence_ids에 없습니다.",
)
if _recorded_draft_path(data_dir, question.get("draft_file")) is None:
_add_issue(
issues,
"BLOCK",
"DRAFT_PATH",
f"{question_location}.draft_file은 .cover-letter/drafts 안의 경로여야 합니다.",
)
verification = question.get("verification")
if verification is not None and not _is_object(verification):
_add_issue(
issues,
"BLOCK",
"VERIFICATION_TYPE",
f"{question_location}.verification은 객체여야 합니다.",
)
if question.get("status") == "user_approved":
if not _is_object(verification) or any(
verification.get(flag) is not True for flag in REQUIRED_VERIFICATION_FLAGS
):
_add_issue(
issues,
"BLOCK",
"VERIFICATION_INCOMPLETE",
f"{question_location}의 최종 확인 항목이 모두 true여야 합니다.",
)
for digest_name in ("draft_sha256", "context_sha256"):
digest = (
verification.get(digest_name) if _is_object(verification) else None
)
if (
not isinstance(digest, str)
or re.fullmatch(r"[0-9a-f]{64}", digest) is None
):
_add_issue(
issues,
"BLOCK",
"VERIFICATION_DIGEST",
f"{question_location}.verification.{digest_name}이 필요합니다.",
)
for ref, location in story_references:
if ref not in story_ids:
_add_issue(
issues,
"BLOCK",
"UNKNOWN_STORY",
f"{location}가 없는 story id {_safe_reference_label(ref)}를 참조합니다.",
)
for ref, location in value_references:
if ref not in value_ids:
_add_issue(
issues,
"BLOCK",
"UNKNOWN_VALUE",
f"{location}가 없는 value id {_safe_reference_label(ref)}를 참조합니다.",
)
for ref in motivation_references:
if (ref.startswith("E") and ref not in story_ids) or (
ref.startswith("V") and ref not in value_ids
):
_add_issue(
issues,
"BLOCK",
"MOTIVATION_EVIDENCE",
f"motivation.evidence_ids가 없는 ID {_safe_reference_label(ref)}를 참조합니다.",
)
elif ref not in eligible_motivation_ids:
_add_issue(
issues,
"BLOCK",
"MOTIVATION_EVIDENCE",
"확정 motivation은 confirmed/locked이면서 use_permission=allowed인 "
f"근거만 참조할 수 있습니다: {_safe_reference_label(ref)}",
)
allowed_outline_ids = story_ids | evidence_ids | value_ids | app_ids | {
identifier
for identifier, location in identifiers.items()
if location.startswith("applications[") and ".requirements[" in location
}
for ref, location in outline_references:
if ref not in allowed_outline_ids or VALID_EVIDENCE_ID.fullmatch(ref) is None:
_add_issue(
issues,
"BLOCK",
"UNKNOWN_OUTLINE_EVIDENCE",
f"{location}가 사용할 수 없는 evidence id {_safe_reference_label(ref)}를 참조합니다.",
)
if _is_object(session):
active_app = session.get("active_application_id")
active_question = session.get("active_question_id")
if isinstance(active_app, str) and active_app and active_app not in app_ids:
_add_issue(
issues,
"BLOCK",
"UNKNOWN_ACTIVE_APPLICATION",
f"활성 application id {_safe_reference_label(active_app)}가 없습니다.",
)
if (
isinstance(active_question, str)
and active_question
and active_question not in question_ids
):
_add_issue(
issues,
"BLOCK",
"UNKNOWN_ACTIVE_QUESTION",
f"활성 question id {_safe_reference_label(active_question)}가 없습니다.",
)
if (
isinstance(active_app, str)
and isinstance(active_question, str)
and active_app
and active_question
and question_parent.get(active_question) != active_app
):
_add_issue(
issues,
"BLOCK",
"ACTIVE_QUESTION_MISMATCH",
"활성 question이 활성 application에 속하지 않습니다.",
)
status = max((item["status"] for item in issues), key=SEVERITY.get, default="PASS")
if not issues:
_add_issue(issues, "PASS", "SCHEMA_VALID", "워크스페이스 스키마와 참조가 유효합니다.")
return {"status": status, "workspace": str(data_dir), "issues": issues}
def _eligible_story(story: Mapping[str, Any]) -> bool:
return _is_enum(story.get("status"), {"confirmed", "locked"}) and story.get(
"use_permission"
) == "allowed"
def _find_application(documents: Mapping[str, Any], application_id: str) -> dict[str, Any] | None:
applications = documents.get("applications.json", {}).get("applications", [])
return next(
(item for item in applications if _is_object(item) and item.get("id") == application_id),
None,
)
def _find_question(application: Mapping[str, Any], question_id: str) -> dict[str, Any] | None:
return next(
(
item
for item in application.get("questions", [])
if _is_object(item) and item.get("id") == question_id
),
None,
)
def _approval_is_current(
data_dir: Path,
application: Mapping[str, Any],
question: Mapping[str, Any],
) -> bool:
verification = question.get("verification", {})
if not _is_object(verification):
return False
source = _recorded_draft_path(data_dir, question.get("draft_file"))
if source is None:
return False
try:
text = source.read_text(encoding="utf-8")
context = build_context(
data_dir,
str(application.get("id", "")),
str(question.get("id", "")),
)
except (HarnessError, OSError, UnicodeError):
return False
return (
verification.get("draft_sha256")
== hashlib.sha256(text.encode("utf-8")).hexdigest()
and verification.get("context_sha256") == _json_digest(context)
)
def workspace_status(data_dir: Path) -> dict[str, Any]:
validation = validate_workspace(data_dir)
if validation["status"] == "BLOCK":
return {
"status": "BLOCK",
"workspace": str(data_dir),
"stage": "INVALID",
"ready_for_context": False,
"ready_for_export": False,
"approval_current": False,
"counts": {},
"issues": validation["issues"],
}
documents = _load_workspace(data_dir)
profile = documents["profile.json"]
stories = documents["stories.json"].get("stories", [])
applications = documents["applications.json"].get("applications", [])
session = documents["session.json"]
eligible_stories = [story for story in stories if _is_object(story) and _eligible_story(story)]
confirmed_apps = [app for app in applications if _is_object(app) and app.get("confirmed") is True]
active_app = _find_application(documents, session.get("active_application_id", ""))
active_question = (
_find_question(active_app, session.get("active_question_id", "")) if active_app else None
)
selected = set(active_question.get("story_ids", [])) if active_question else set()
eligible_ids = {story.get("id") for story in eligible_stories}
selection_ready = bool(selected) and selected <= eligible_ids
context_ready = selection_ready
if (
active_app is not None
and active_question is not None
and active_app.get("confirmed") is True
and active_question.get("outline_approved") is True
and selection_ready
):
try:
build_context(
data_dir,
str(active_app.get("id", "")),
str(active_question.get("id", "")),
)
except HarnessError:
context_ready = False
candidate = profile.get("candidate", {})
scope_ready = _is_object(candidate) and _has_text(candidate.get("career_level")) and _has_text_items(
candidate.get("target_roles")
)
if profile.get("confirmed") is not True:
stage = "SCOPED" if scope_ready else "NEW"
elif not eligible_stories:
stage = "COLLECTING"
elif not confirmed_apps:
stage = "PERSONAL_MODEL_CONFIRMED"
elif active_app is None or active_question is None:
stage = "TARGET_READY"
elif active_app.get("confirmed") is not True:
stage = "PERSONAL_MODEL_CONFIRMED"
elif active_question.get("outline_approved") is not True:
stage = "QUESTION_READY"
elif not context_ready:
stage = "NEEDS_RECHECK"
else:
question_status = active_question.get("status")
stage = {
"drafted": "DRAFTED",
"verified": "VERIFIED",
"user_approved": "USER_APPROVED",
"needs_recheck": "NEEDS_RECHECK",
}.get(question_status, "OUTLINE_APPROVED")
verification = active_question.get("verification", {}) if active_question else {}
verification_complete = (
_is_object(verification)
and all(verification.get(flag) is True for flag in REQUIRED_VERIFICATION_FLAGS)
)
approval_current = bool(
active_app
and active_question
and stage == "USER_APPROVED"
and verification_complete
and _approval_is_current(data_dir, active_app, active_question)
)
if stage == "USER_APPROVED" and not approval_current:
stage = "NEEDS_RECHECK"
ready_for_context = context_ready and stage in {
"OUTLINE_APPROVED",
"DRAFTED",
"VERIFIED",
"USER_APPROVED",
"NEEDS_RECHECK",
}
ready_for_export = stage == "USER_APPROVED" and approval_current
overall_status = (
"PASS"
if ready_for_context and stage not in {"NEEDS_RECHECK"}
else "WARN"
)
return {
"status": overall_status,
"workspace": str(data_dir),
"stage": stage,
"ready_for_context": ready_for_context,
"ready_for_export": ready_for_export,
"approval_current": approval_current,
"active_application_id": session.get("active_application_id", ""),
"active_question_id": session.get("active_question_id", ""),
"counts": {
"values": len(profile.get("values", [])),
"stories": len(stories),
"eligible_stories": len(eligible_stories),
"applications": len(applications),
"confirmed_applications": len(confirmed_apps),
},
"issues": [],
}
NEXT_MESSAGES = {
"INVALID": "`./cover-letter validate`의 차단 항목을 먼저 수정하세요.",
"NEW": "AI 대화에서 `자소서: 시작`으로 직무·경력 단계·민감정보 기준을 정하세요.",
"SCOPED": "AI 대화에서 `자소서: 인터뷰 경험`으로 실제 경험 하나를 구체화하세요.",
"COLLECTING": "AI 대화에서 `자소서: 확인 E###`로 본인 역할·결과·사용 허용 범위를 확인하세요.",
"PERSONAL_MODEL_CONFIRMED": "AI 대화에서 `자소서: 지원처 등록`으로 출처 있는 공고를 등록하세요.",
"TARGET_READY": "AI 대화에서 `자소서: 문항 추가 A###`로 문항과 분량 규칙을 등록하세요.",
"QUESTION_READY": "AI 대화에서 `자소서: 개요 Q###`로 근거 ID가 있는 개요를 승인하세요.",
"OUTLINE_APPROVED": "아래 context 명령으로 승인된 재료만 꺼내 근거 주석이 있는 초안을 작성하세요.",
"DRAFTED": "아래 check-draft 명령으로 사실·역할·분량·개인정보를 검증하세요.",
"VERIFIED": "여섯 확인 항목을 직접 읽은 뒤 아래 approve 명령으로 현재 버전을 잠그세요.",
"USER_APPROVED": "아래 export 명령으로 주석을 제거한 별도 제출 파일을 만드세요.",
"NEEDS_RECHECK": "변경된 초안을 check-draft로 확인하고 approve를 다시 실행하세요.",
}
def _next_exact_commands(data_dir: Path, report: Mapping[str, Any]) -> list[str]:
stage = str(report.get("stage", ""))
application_id = str(report.get("active_application_id", ""))
question_id = str(report.get("active_question_id", ""))
if stage == "OUTLINE_APPROVED" and application_id and question_id:
return [
"./cover-letter context "
f"--application {shlex.quote(application_id)} --question {shlex.quote(question_id)}"
]
try:
documents = _load_workspace(data_dir)
application = _find_application(documents, application_id)
question = _find_question(application, question_id) if application else None
source = (
_recorded_draft_path(data_dir, question.get("draft_file")) if question else None
)
except HarnessError:
source = None
if source is None:
return []
quoted_source = shlex.quote(str(source))
if stage == "DRAFTED":
return [f"./cover-letter check-draft {quoted_source}"]
if stage == "VERIFIED":
return [f"./cover-letter approve {quoted_source} --confirm-all"]
if stage == "NEEDS_RECHECK":
return [
f"./cover-letter check-draft {quoted_source}",
f"./cover-letter approve {quoted_source} --confirm-all",
]
if stage == "USER_APPROVED":
destination = data_dir / "exports" / f"{source.stem}.txt"
return [
f"./cover-letter export {quoted_source} {shlex.quote(str(destination))}"
]
return []
def _eligible_evidence(item: Mapping[str, Any], story: Mapping[str, Any]) -> bool:
if "verified" in item:
verified = item.get("verified") is True
elif "status" in item:
verified = item.get("status") in {"confirmed", "locked"}
else:
verified = story.get("status") in {"confirmed", "locked"}
permission = item.get("use_permission", story.get("use_permission", "ask"))
return verified and permission == "allowed" and _eligible_story(story)
def _searchable_evidence_text(value: Any, field: str = "") -> str:
"""Flatten claim material while excluding IDs that contain incidental digits."""
if field == "id" or field.endswith("_ids"):
return ""
if isinstance(value, Mapping):
return " ".join(_searchable_evidence_text(item, str(key)) for key, item in value.items())
if isinstance(value, list):
return " ".join(_searchable_evidence_text(item, field) for item in value)
if isinstance(value, (str, int, float)) and not isinstance(value, bool):
return str(value)
return ""
def _expanded_period_text(value: Any) -> str:
if not isinstance(value, str):
return ""
expanded = [value]
match = re.fullmatch(r"\s*(\d{4})-(\d{1,2})(?:-(\d{1,2}))?\s*", value)
if match:
year, month, day = match.groups()
expanded.extend((f"{year}년", f"{int(month)}월"))
if day is not None:
expanded.append(f"{int(day)}일")
return " ".join(expanded)
def _copy_fields(value: Mapping[str, Any], fields: Sequence[str]) -> dict[str, Any]:
return {field: copy.deepcopy(value[field]) for field in fields if field in value}
def _project_record(
value: Mapping[str, Any],
*,
text_fields: Sequence[str] = (),
text_list_fields: Sequence[str] = (),
bool_fields: Sequence[str] = (),
integer_fields: Sequence[str] = (),
) -> dict[str, Any]:
"""Copy only documented scalar/list fields so extra nested notes never enter context."""
projected: dict[str, Any] = {}
for field in text_fields:
if isinstance(value.get(field), str):
projected[field] = value[field]
for field in text_list_fields:
raw_items = value.get(field)
if _is_list(raw_items):
projected[field] = [item for item in raw_items if isinstance(item, str)]
for field in bool_fields:
if isinstance(value.get(field), bool):
projected[field] = value[field]
for field in integer_fields:
raw_integer = value.get(field)
if isinstance(raw_integer, int) and not isinstance(raw_integer, bool):
projected[field] = raw_integer
return projected
def _project_outcomes(value: Any) -> dict[str, Any]:
if not _is_object(value):
return {}
return _project_record(
value,
text_fields=("attribution", "owner_scope"),
text_list_fields=("measured", "observed"),
)
def _project_evidence(value: Mapping[str, Any]) -> dict[str, Any]:
return _project_record(
value,
text_fields=("id", "claim", "source", "status", "use_permission", "owner_scope"),
bool_fields=("verified",),
)
def _project_value(value: Mapping[str, Any]) -> dict[str, Any]:
return _project_record(
value,
text_fields=("id", "statement", "origin", "behavior", "status", "use_permission"),
text_list_fields=("story_ids",),
)
def _project_story(value: Mapping[str, Any]) -> dict[str, Any]:
projected = _project_record(
value,
text_fields=(
"id",
"title",
"period",
"context",
"user_role",
"team_role",
"reflection",
"status",
"use_permission",
),
text_list_fields=(
"constraints",
"decisions",
"actions",
"alternatives",
"skills",
"value_ids",
),
)
projected["outcomes"] = _project_outcomes(value.get("outcomes"))
return projected
def _project_posting(value: Any) -> dict[str, Any]:
if not _is_object(value):
return {}
return _project_record(value, text_fields=POSTING_CONTEXT_FIELDS)
def _project_requirement(value: Mapping[str, Any]) -> dict[str, Any]:
return _project_record(
value,
text_fields=("id", "text", "priority"),
text_list_fields=("story_ids",),
)
def _project_outline(value: Any) -> dict[str, Any]:
if not _is_object(value):
return {}
return _project_record(
value,
text_fields=("thesis",),
text_list_fields=(
"beats",
"evidence_ids",
"paragraphs",
"personal_fingerprints",
"excluded_items",
),
integer_fields=("estimated_length",),
)
def _url_has_sensitive_material(url: str) -> bool:
candidate = unquote(url.rstrip(".,;:!?)]}'\"。!?、,;:")).casefold()
query_and_fragment = re.split(r"[?#]", candidate, maxsplit=1)
if len(query_and_fragment) == 1:
return False
tail = query_and_fragment[1]
for part in re.split(r"[&#;]", tail):
key = part.split("=", 1)[0].strip().replace("-", "_")
normalized_keys = {item.replace("-", "_") for item in SENSITIVE_URL_KEYS}
if key in normalized_keys or key.endswith("_code") or any(
marker in key for marker in ("token", "secret", "signature", "credential", "password")
):
return True
return False
def _is_private_bare_endpoint(value: str) -> bool:
authority = value.split("/", 1)[0]
if authority.startswith("[") and "]" in authority:
host = authority[1 : authority.index("]")]
elif authority.count(":") >= 2:
host = authority
else:
host = authority.split(":", 1)[0]
host = host.lower().rstrip(".")
if host == "localhost" or host.endswith((".internal", ".local", ".lan")):
return True
try:
return not ipaddress.ip_address(host).is_global
except ValueError:
return "." not in host and bool(re.fullmatch(r"[a-z][a-z0-9-]{0,62}", host))
def _is_private_url(url: str) -> bool:
candidate = url.rstrip(".,;:!?)]}'\"。!?、,;:")
if _url_has_sensitive_material(candidate):
return True
try:
if "://" not in candidate:
return True
authority_and_path = candidate.rsplit("://", 1)[1]
parsed = urlsplit("scope://" + authority_and_path)
hostname = (parsed.hostname or "").lower().rstrip(".")
except (UnicodeError, ValueError):
return True
if not hostname or parsed.username or parsed.password:
return True
if hostname == "localhost" or hostname.endswith((".local", ".internal", ".lan")):
return True
try:
address = ipaddress.ip_address(hostname)
except ValueError:
try:
ascii_hostname = hostname.encode("idna").decode("ascii")
except UnicodeError:
return True
labels = ascii_hostname.split(".")
return len(labels) < 2 or any(
re.fullmatch(r"[a-z0-9](?:[a-z0-9-]{0,61}[a-z0-9])?", label)
is None
for label in labels
)
return not address.is_global
def _redact_context(value: Any, blocked_terms: Sequence[str]) -> Any:
"""Remove obvious sensitive strings without returning the matched value."""
if isinstance(value, str):
redacted = unicodedata.normalize("NFC", value)
redacted = "".join(
character
for character in redacted
if unicodedata.category(character) != "Cf"
and not (
unicodedata.category(character) == "Cc"
and character not in "\n\r\t"
)
)
for pattern in SENSITIVE_PATTERNS.values():
redacted = pattern.sub("[민감정보 제거]", redacted)
redacted = URL_PATTERN.sub(
lambda match: "[내부 URL 제거]" if _is_private_url(match.group(0)) else match.group(0),
redacted,
)
redacted = BARE_ENDPOINT_PATTERN.sub(
lambda match: (
"[내부 엔드포인트 제거]"
if _is_private_bare_endpoint(match.group(0))
else match.group(0)
),
redacted,
)
for term in blocked_terms:
if term:
redacted = redacted.replace(term, "[사용 금지 정보 제거]")
return redacted
if isinstance(value, list):
return [_redact_context(item, blocked_terms) for item in value]
if isinstance(value, Mapping):
safe_mapping: dict[Any, Any] = {}
for key, item in value.items():
if isinstance(key, str) and _contains_sensitive_text(key, blocked_terms):
continue
safe_mapping[key] = _redact_context(item, blocked_terms)
return safe_mapping
return copy.deepcopy(value)
def _contains_sensitive_text(text: str, blocked_terms: Sequence[str]) -> bool:
return any(pattern.search(text) for pattern in SENSITIVE_PATTERNS.values()) or any(
term in text for term in blocked_terms if term
) or any(_is_private_url(match.group(0)) for match in URL_PATTERN.finditer(text)) or any(
_is_private_bare_endpoint(match.group(0))
for match in BARE_ENDPOINT_PATTERN.finditer(text)
)
def build_context(data_dir: Path, application_id: str, question_id: str) -> dict[str, Any]:
validation = validate_workspace(data_dir)
if validation["status"] == "BLOCK":
messages = "; ".join(item["message"] for item in validation["issues"] if item["status"] == "BLOCK")
raise HarnessError(f"워크스페이스 검증이 필요합니다: {messages}")
documents = _load_workspace(data_dir)
profile = documents["profile.json"]
if profile.get("confirmed") is not True:
raise HarnessError("profile.json이 아직 확인되지 않았습니다(confirmed=false).")
application = _find_application(documents, application_id)
if application is None:
raise HarnessError(f"application {application_id}를 찾을 수 없습니다.")
if application.get("confirmed") is not True:
raise HarnessError(f"application {application_id}가 아직 확인되지 않았습니다.")
question = _find_question(application, question_id)
if question is None:
raise HarnessError(f"application {application_id}에 question {question_id}가 없습니다.")
if question.get("outline_approved") is not True:
raise HarnessError(f"question {question_id}의 개요가 아직 승인되지 않았습니다.")
outline = question.get("outline", {})
approved_evidence_ids = set(outline.get("evidence_ids", [])) if _is_object(outline) else set()
if not approved_evidence_ids:
raise HarnessError(f"question {question_id}의 승인 개요에 evidence_ids가 없습니다.")
selected_ids = question.get("story_ids", [])
if not isinstance(selected_ids, list) or not selected_ids:
raise HarnessError(f"question {question_id}에 사용할 story_ids가 없습니다.")
story_map = {
story.get("id"): story
for story in documents["stories.json"].get("stories", [])
if _is_object(story) and isinstance(story.get("id"), str)
}
confirmed_values = {
value.get("id"): value
for value in profile.get("values", [])
if _is_object(value)
and isinstance(value.get("id"), str)
and _is_enum(value.get("status"), {"confirmed", "locked"})
and value.get("use_permission") == "allowed"
}
available_ids = set(selected_ids) | set(confirmed_values)
for story_id in selected_ids:
story = story_map.get(story_id)
if _is_object(story):
available_ids.update(
item.get("id")
for item in story.get("evidence", [])
if _is_object(item)
and isinstance(item.get("id"), str)
and _eligible_evidence(item, story)
)
available_ids.add(application_id)
available_ids.update(
requirement.get("id")
for requirement in application.get("requirements", [])
if _is_object(requirement) and isinstance(requirement.get("id"), str)
)
unavailable = sorted(approved_evidence_ids - available_ids)
if unavailable:
raise HarnessError(
"승인 개요가 현재 문항에서 사용할 수 없거나 미확인인 근거를 참조합니다: "
+ ", ".join(unavailable)
)
catalog, catalog_available, catalog_error = _evidence_catalog(data_dir)
if not catalog_available:
raise HarnessError(catalog_error or "근거 카탈로그를 읽을 수 없습니다.")
ineligible = sorted(
evidence_id
for evidence_id in approved_evidence_ids
if evidence_id not in catalog
or catalog[evidence_id].get("verified") is not True
or catalog[evidence_id].get("allowed") is not True
)
if ineligible:
raise HarnessError("승인 개요에 미확인·비공개 근거가 있습니다: " + ", ".join(ineligible))
selected_stories: list[dict[str, Any]] = []
for story_id in selected_ids:
story = story_map.get(story_id)
if story is None:
raise HarnessError(f"선택한 story {story_id}가 없습니다.")
if not _eligible_story(story):
raise HarnessError(
f"story {story_id}는 confirmed/locked 상태이면서 use_permission=allowed여야 합니다."
)
safe_story = _project_story(story)
evidence = story.get("evidence", [])
safe_story["evidence"] = [
_project_evidence(item)
for item in evidence
if _is_object(item)
and item.get("id") in approved_evidence_ids
and _eligible_evidence(item, story)
]
selected_stories.append(safe_story)
candidate = profile.get("candidate", {})
safe_candidate = (
_project_record(
candidate,
text_fields=("career_level", "current_status"),
text_list_fields=("target_roles",),
)
if _is_object(candidate)
else {}
)
motivation = candidate.get("motivation", {}) if _is_object(candidate) else {}
if _is_object(motivation):
approved_motivation_ids = [
evidence_id
for evidence_id in motivation.get("evidence_ids", [])
if isinstance(evidence_id, str) and evidence_id in approved_evidence_ids
]
if approved_motivation_ids:
safe_motivation = _project_record(
motivation,
text_fields=("why_it", "why_role", "future_direction"),
)
safe_motivation["evidence_ids"] = approved_motivation_ids
safe_candidate["motivation"] = safe_motivation
voice = profile.get("voice", {})
safe_voice = (
_project_record(
voice,
text_list_fields=("preferred_tone", "avoid_phrases"),
)
if _is_object(voice)
else {}
)
safe_voice["samples"] = [
_project_record(
sample,
text_fields=("id", "kind", "text", "use_permission"),
bool_fields=("user_authored",),
)
for sample in (voice.get("samples", []) if _is_object(voice) else [])
if _is_object(sample)
and sample.get("user_authored") is True
and sample.get("use_permission") == "allowed"
]
safe_profile = {
"candidate": safe_candidate,
"values": [
_project_value(value)
for value in profile.get("values", [])
if _is_object(value)
and value.get("id") in approved_evidence_ids
and _is_enum(value.get("status"), {"confirmed", "locked"})
and value.get("use_permission") == "allowed"
],
"voice": safe_voice,
"privacy_rules": {
"redaction_rule_count": len(profile.get("privacy", {}).get("redactions", [])),
"excluded_item_count": len(profile.get("privacy", {}).get("do_not_use", [])),
},
"confirmed": True,
}
safe_application = _copy_fields(application, ("id", "company", "role", "confirmed"))
if application_id in approved_evidence_ids:
safe_application["posting"] = _project_posting(application.get("posting"))
safe_application["requirements"] = [
_project_requirement(requirement)
for requirement in application.get("requirements", [])
if _is_object(requirement) and requirement.get("id") in approved_evidence_ids
]
safe_question = _project_record(
question,
text_fields=("id", "prompt", "required_format", "draft_file"),
text_list_fields=("story_ids",),
bool_fields=("count_spaces",),
integer_fields=("character_limit", "character_minimum"),
)
context = {
"schema_version": SCHEMA_VERSION,
"profile": safe_profile,
"application": safe_application,
"question": safe_question,
"approved_outline": _project_outline(outline),
"stories": selected_stories,
"evidence_annotation_format": "<!-- evidence: E001,V001 -->",
}
blocked_terms = [
term
for term in profile.get("privacy", {}).get("do_not_use", [])
if isinstance(term, str) and term
]
return _redact_context(context, blocked_terms)
def _evidence_catalog(data_dir: Path | None) -> tuple[dict[str, dict[str, Any]], bool, str | None]:
if data_dir is None:
return {}, False, "워크스페이스를 찾지 못했습니다. --workspace로 지정하세요."
try:
documents = _load_workspace(data_dir)
except HarnessError as exc:
return {}, False, str(exc)
catalog: dict[str, dict[str, Any]] = {}
profile = documents["profile.json"]
for value in profile.get("values", []):
if not _is_object(value) or not isinstance(value.get("id"), str):
continue
projected_value = _project_value(value)
catalog[value["id"]] = {
"verified": profile.get("confirmed") is True
and _is_enum(value.get("status"), {"confirmed", "locked"}),
"allowed": value.get("use_permission") == "allowed",
"kind": "value",
"text": _searchable_evidence_text(projected_value),
"numeric_text": _searchable_evidence_text(projected_value),
"period_text": "",
"ownership_text": _searchable_evidence_text(projected_value),
"result_text": "",
"entity_terms": [],
"story_id": None,
"result_owner_scope": None,
}
for story in documents["stories.json"].get("stories", []):
if not _is_object(story):
continue
story_id = story.get("id")
story_source = _project_story(story)
story_numeric_source = copy.deepcopy(story_source)
story_numeric_source.pop("period", None)
ownership_source = _copy_fields(story, STORY_OWNERSHIP_FIELDS)
outcomes = story.get("outcomes", {})
if isinstance(story_id, str):
catalog[story_id] = {
"verified": _is_enum(story.get("status"), {"confirmed", "locked"}),
"allowed": story.get("use_permission") == "allowed",
"kind": "story",
"text": _searchable_evidence_text(story_source),
"numeric_text": _searchable_evidence_text(story_numeric_source),
"period_text": _expanded_period_text(story.get("period")),
"ownership_text": _searchable_evidence_text(ownership_source),
"result_text": _searchable_evidence_text(_project_outcomes(outcomes)),
"entity_terms": list(_korean_specific_terms(str(story.get("title", "")))),
"story_id": story_id,
"result_owner_scope": (
outcomes.get("owner_scope") if _is_object(outcomes) else None
),
}
for item in story.get("evidence", []):
if not _is_object(item) or not isinstance(item.get("id"), str):
continue
if "verified" in item:
verified = item.get("verified") is True
elif "status" in item:
verified = _is_enum(item.get("status"), {"confirmed", "locked"})
else:
verified = _is_enum(story.get("status"), {"confirmed", "locked"})
permission = item.get("use_permission", story.get("use_permission", "ask"))
item_owner_scope = item.get(
"owner_scope",
outcomes.get("owner_scope") if _is_object(outcomes) else None,
)
projected_item = _project_evidence(item)
catalog[item["id"]] = {
"verified": verified
and _is_enum(story.get("status"), {"confirmed", "locked"}),
"allowed": permission == "allowed" and story.get("use_permission") == "allowed",
"kind": "evidence",
"text": _searchable_evidence_text(projected_item),
"numeric_text": _searchable_evidence_text(projected_item),
"period_text": "",
"ownership_text": (
_searchable_evidence_text(projected_item)
if item_owner_scope == "user"
else ""
),
"result_text": _searchable_evidence_text(projected_item),
"entity_terms": [],
"story_id": story_id,
"result_owner_scope": item_owner_scope,
}
candidate = profile.get("candidate", {})
motivation = candidate.get("motivation", {}) if _is_object(candidate) else {}
if _is_object(motivation) and profile.get("confirmed") is True:
motivation_text = " ".join(
str(motivation.get(field, ""))
for field in ("why_it", "why_role", "future_direction")
if isinstance(motivation.get(field), str)
)
for evidence_id in motivation.get("evidence_ids", []):
if (
isinstance(evidence_id, str)
and evidence_id in catalog
and catalog[evidence_id].get("verified") is True
and catalog[evidence_id].get("allowed") is True
):
catalog[evidence_id]["text"] = (
str(catalog[evidence_id].get("text", "")) + " " + motivation_text
).strip()
for application in documents["applications.json"].get("applications", []):
if not _is_object(application):
continue
application_id = application.get("id")
application_source = {
"company": application.get("company", ""),
"role": application.get("role", ""),
"posting": _project_posting(application.get("posting")),
}
if isinstance(application_id, str):
catalog[application_id] = {
"verified": application.get("confirmed") is True,
"allowed": application.get("confirmed") is True,
"kind": "application",
"text": _searchable_evidence_text(application_source),
"numeric_text": _searchable_evidence_text(application_source),
"period_text": "",
"ownership_text": _searchable_evidence_text(application_source),
"result_text": _searchable_evidence_text(application_source),
"entity_terms": list(
_korean_specific_terms(
f"{application.get('company', '')} {application.get('role', '')}"
)
),
"story_id": None,
"result_owner_scope": None,
}
for requirement in application.get("requirements", []):
if not _is_object(requirement) or not isinstance(requirement.get("id"), str):
continue
projected_requirement = _project_requirement(requirement)
catalog[requirement["id"]] = {
"verified": application.get("confirmed") is True,
"allowed": application.get("confirmed") is True,
"kind": "requirement",
"text": _searchable_evidence_text(projected_requirement),
"numeric_text": _searchable_evidence_text(projected_requirement),
"period_text": "",
"ownership_text": _searchable_evidence_text(projected_requirement),
"result_text": _searchable_evidence_text(projected_requirement),
"entity_terms": [],
"story_id": None,
"result_owner_scope": None,
}
return catalog, True, None
def _privacy_terms(data_dir: Path | None) -> list[str]:
if data_dir is None:
return []
try:
profile = _read_json(data_dir / "profile.json")
except HarnessError:
return []
terms = profile.get("privacy", {}).get("do_not_use", []) if _is_object(profile) else []
if not _is_list(terms):
return []
return [term for term in terms if isinstance(term, str) and len(term.strip()) >= 2]
def _parse_comments(text: str) -> tuple[list[dict[str, Any]], list[dict[str, Any]]]:
valid: list[dict[str, Any]] = []
valid_spans: set[tuple[int, int]] = set()
for match in VALID_COMMENT.finditer(text):
raw_ids = match.group("ids")
ids = tuple(item.strip().upper() for item in raw_ids.split(",") if item.strip())
valid.append(
{
"start": match.start(),
"end": match.end(),
"line": text.count("\n", 0, match.start()) + 1,
"ids": ids,
"raw": match.group(0),
}
)
valid_spans.add((match.start(), match.end()))
malformed: list[dict[str, Any]] = []
for match in EVIDENCE_LIKE_COMMENT.finditer(text):
if (match.start(), match.end()) not in valid_spans:
malformed.append(
{
"line": text.count("\n", 0, match.start()) + 1,
"raw": match.group(0),
}
)
return valid, malformed
def _mask_comments(text: str) -> str:
return ANY_HTML_COMMENT.sub(lambda match: " " * (match.end() - match.start()), text)
def _sentences(text: str, comments: Sequence[Mapping[str, Any]]) -> list[Sentence]:
masked = _mask_comments(text)
spans: list[tuple[int, int, int]] = []
for match in re.finditer(
r"[^!?。!?\n]+?(?:[!?。!?]+|[.]+(?=\s|$)|(?=\n)|$)", masked
):
raw = match.group(0)
nonspace = re.search(r"\S", raw)
if nonspace is None:
continue
start = match.start() + nonspace.start()
end = match.end()
while end > start and masked[end - 1].isspace():
end -= 1
spans.append((start, end, text.count("\n", 0, start) + 1))
result: list[Sentence] = []
for index, (start, end, line) in enumerate(spans):
paragraph_start = 0
for boundary in re.finditer(r"\n[ \t]*\n", masked[:start]):
paragraph_start = boundary.end()
following_boundary = re.search(r"\n[ \t]*\n", masked[end:])
paragraph_end = (
end + following_boundary.start() if following_boundary is not None else len(text)
)
associated: list[str] = []
for comment in comments:
position = int(comment["start"])
if (
paragraph_start <= position < paragraph_end
and int(comment["end"]) <= start
):
associated.extend(str(item) for item in comment["ids"])
result.append(
Sentence(
text=masked[start:end].strip(),
start=start,
end=end,
line=line,
evidence_ids=tuple(dict.fromkeys(associated)),
)
)
return result
def _check(status: str, code: str, message: str, details: Iterable[Any] = ()) -> dict[str, Any]:
return {
"status": status,
"code": code,
"message": message,
"details": list(details),
}
def _submission_text(text: str) -> str:
visible = STANDALONE_VALID_COMMENT_LINE.sub("", text)
visible = unicodedata.normalize("NFC", VALID_COMMENT.sub("", visible))
visible = re.sub(r"[ \t]+(?=\n|$)", "", visible)
visible = visible.strip()
return visible + "\n" if visible else ""
def _visible_character_count(text: str, count_spaces: bool) -> int:
visible = _submission_text(text).rstrip("\n")
if count_spaces:
return len(visible)
return len(re.sub(r"\s+", "", visible))
def _recorded_draft_path(data_dir: Path, raw_path: Any) -> Path | None:
if not isinstance(raw_path, str) or not raw_path.strip():
return None
path = Path(raw_path).expanduser()
if not path.is_absolute():
path = data_dir.parent / path if path.parts and path.parts[0] == DATA_DIRECTORY else data_dir / path
try:
resolved = path.resolve()
resolved.relative_to((data_dir / "drafts").resolve())
except (ValueError, OSError, RuntimeError):
return None
return resolved
def _draft_binding(
data_dir: Path, source: Path, *, required: bool = False
) -> tuple[dict[str, Any], dict[str, Any]] | None:
documents = _load_workspace(data_dir)
matches: list[tuple[dict[str, Any], dict[str, Any]]] = []
resolved_source = _resolve_path(source, "초안 경로")
for application in documents["applications.json"].get("applications", []):
if not _is_object(application):
continue
for question in application.get("questions", []):
if not _is_object(question):
continue
recorded = _recorded_draft_path(data_dir, question.get("draft_file"))
if recorded == resolved_source:
matches.append((application, question))
if len(matches) > 1:
raise HarnessError("초안 파일이 둘 이상의 문항에 연결되어 있습니다.")
if not matches:
if required:
raise HarnessError("초안 파일을 참조하는 문항이 없습니다. question.draft_file을 확인하세요.")
return None
return matches[0]
def _question_constraints(question: Mapping[str, Any]) -> tuple[int | None, int | None, bool, str | None]:
limit = question.get("character_limit")
if not isinstance(limit, int) or isinstance(limit, bool) or limit <= 0:
limit = None
minimum = question.get("character_minimum")
if not isinstance(minimum, int) or isinstance(minimum, bool) or minimum < 0:
minimum = None
count_spaces = question.get("count_spaces", True) is not False
required_format = question.get("required_format")
if required_format not in {"plain_text", "markdown"}:
required_format = None
return limit, minimum, count_spaces, required_format
def _inferred_constraints(
data_dir: Path | None, source: Path | None = None
) -> tuple[int | None, int | None, bool, str | None]:
if data_dir is None:
return None, None, True, None
try:
documents = _load_workspace(data_dir)
except HarnessError:
return None, None, True, None
if source is not None:
binding = _draft_binding(data_dir, source)
if binding is not None:
return _question_constraints(binding[1])
session = documents["session.json"]
app = _find_application(documents, session.get("active_application_id", ""))
question = _find_question(app, session.get("active_question_id", "")) if app else None
if not question:
return None, None, True, None
return _question_constraints(question)
def _korean_specific_terms(text: str) -> set[str]:
terms: set[str] = set()
for raw_token in KOREAN_TERM_PATTERN.findall(text):
token = raw_token
canonical = next(
(
name
for name, aliases in KOREAN_TERM_ALIASES.items()
if any(token.startswith(alias) for alias in aliases)
),
None,
)
if canonical is not None:
terms.add(canonical)
continue
for particle in KOREAN_PARTICLES:
if token.endswith(particle) and len(token) - len(particle) >= 2:
token = token[: -len(particle)]
break
canonical = next(
(
name
for name, aliases in KOREAN_TERM_ALIASES.items()
if any(token.startswith(alias) for alias in aliases)
),
None,
)
if canonical is not None:
terms.add(canonical)
continue
if len(token) < 2 or any(token.startswith(stem) for stem in GENERIC_KOREAN_STEMS):
continue
terms.add(token)
return terms
def _number_claim_supported(
number: str,
unit: str | None,
records: Sequence[Mapping[str, Any]],
) -> bool:
numeric_text = " ".join(str(record.get("numeric_text", "")) for record in records)
if unit in TIME_NUMBER_UNITS:
numeric_text += " " + " ".join(
str(record.get("period_text", "")) for record in records
)
if unit is None:
return number in NUMBER_PATTERN.findall(numeric_text)
aliases = ("%", "퍼센트") if unit in {"%", "퍼센트"} else (unit,)
particle = r"(?:의|를|을|이|가|은|는|도|만)?"
return any(
re.search(
rf"(?<![A-Za-z0-9]){re.escape(number)}(?![A-Za-z0-9])\s*"
rf"{re.escape(alias)}{particle}(?![가-힣A-Za-z])",
numeric_text,
flags=re.IGNORECASE,
)
is not None
for alias in aliases
)
def _korean_number_claim_supported(
number: str,
unit: str,
records: Sequence[Mapping[str, Any]],
) -> bool:
numeric_text = " ".join(str(record.get("numeric_text", "")) for record in records)
return re.search(
rf"(?<![가-힣]){re.escape(number)}\s*{re.escape(unit)}"
rf"(?:의|를|을|이|가|은|는|도|만)?(?![가-힣])",
numeric_text,
) is not None
def _korean_number_claims(text: str) -> list[Any]:
matches: list[Any] = []
seen_spans: set[tuple[int, int]] = set()
for pattern in KOREAN_NUMBER_CLAIM_PATTERNS:
for match in pattern.finditer(text):
if match.span() not in seen_spans:
seen_spans.add(match.span())
matches.append(match)
return sorted(matches, key=lambda match: match.start())
def _grounding_problems(
sentence: Sentence,
catalog: Mapping[str, Mapping[str, Any]],
blocked_terms: Sequence[str],
) -> list[dict[str, Any]]:
if _contains_sensitive_text(sentence.text, blocked_terms):
return []
records = [
catalog[evidence_id]
for evidence_id in sentence.evidence_ids
if evidence_id in catalog
and catalog[evidence_id].get("verified") is True
and catalog[evidence_id].get("allowed") is True
]
if not records:
return []
personal_records = [record for record in records if record.get("kind") in {"story", "evidence"}]
claim_source_records = personal_records or records
source_text = " ".join(str(record.get("text", "")) for record in claim_source_records)
ownership_text = " ".join(
str(record.get("ownership_text", record.get("text", "")))
for record in claim_source_records
)
sentence_folded = sentence.text.casefold()
source_folded = source_text.casefold()
ownership_folded = ownership_text.casefold()
problems: list[dict[str, Any]] = []
sentence_without_urls = URL_PATTERN.sub("", sentence.text)
sentence_terms = {
token.casefold()
for token in LATIN_TERM_PATTERN.findall(sentence_without_urls)
if token.casefold() not in LATIN_TERM_STOPWORDS
}
source_terms = {token.casefold() for token in LATIN_TERM_PATTERN.findall(source_text)}
missing_latin = sorted(sentence_terms - source_terms)
if missing_latin:
problems.append({"kind": "unseen_latin_term", "terms": missing_latin})
entity_terms = {
term
for record in claim_source_records
for term in record.get("entity_terms", [])
if isinstance(term, str)
}
sentence_korean_terms = _korean_specific_terms(sentence.text) - entity_terms
source_korean_terms = _korean_specific_terms(source_text) - entity_terms
supported_korean_terms = sentence_korean_terms.intersection(source_korean_terms)
missing_korean_terms = sentence_korean_terms - source_korean_terms
if sentence_korean_terms and (
not supported_korean_terms
or (
len(sentence_korean_terms) >= 4
and len(supported_korean_terms) < 2
and len(missing_korean_terms) >= 2
)
):
problems.append(
{
"kind": "insufficient_korean_grounding",
"unsupported_terms": sorted(missing_korean_terms)[:5],
}
)
if RESULT_CLAIM_PATTERN.search(sentence.text):
result_text = " ".join(
str(record.get("result_text", "")) for record in claim_source_records
)
result_terms = _korean_specific_terms(result_text) - entity_terms
result_overlap = sentence_korean_terms.intersection(result_terms)
if not result_terms or not result_overlap:
problems.append({"kind": "unsupported_result_claim"})
for canonical, aliases in TECHNOLOGY_ALIASES.items():
if any(alias.casefold() in sentence_folded for alias in aliases) and not any(
alias.casefold() in source_folded for alias in aliases
):
problems.append({"kind": "unseen_technology", "term": canonical})
claim_evidence_folded = (
source_folded
if TEAM_SUBJECT_PATTERN.search(sentence.text) or INTENTION_PATTERN.search(sentence.text)
else ownership_folded
)
for label, markers in CLAIM_GROUPS.items():
if any(marker in sentence.text for marker in markers["claim"]) and not any(
marker in claim_evidence_folded for marker in markers["evidence"]
):
problems.append({"kind": "unsupported_action_or_ownership", "term": label})
collective_result_claim = any(
marker in sentence.text for marker in COLLECTIVE_RESULT_MARKERS
)
measured_result_claim = (
any(marker in sentence.text for marker in MEASURED_RESULT_MARKERS)
or (
NUMBER_PATTERN.search(sentence.text) is not None
and METRIC_RESULT_PATTERN.search(sentence.text) is not None
)
)
if (
personal_records
and (collective_result_claim or measured_result_claim)
and TEAM_SUBJECT_PATTERN.search(sentence.text) is None
):
supporting_records = [
record
for record in personal_records
if any(
_number_claim_supported(
claim.group("number"), claim.group("unit"), [record]
)
for claim in NUMBER_CLAIM_PATTERN.finditer(sentence.text)
)
or any(
marker in sentence.text and marker in str(record.get("text", ""))
for marker in COLLECTIVE_RESULT_MARKERS
)
]
nested_supporting_records = [
record for record in supporting_records if record.get("kind") == "evidence"
]
scope_records = nested_supporting_records or supporting_records or personal_records
owner_scopes = {
record.get("result_owner_scope")
for record in scope_records
if record.get("result_owner_scope") in {"user", "team", "shared"}
}
if owner_scopes and owner_scopes != {"user"}:
problems.append({"kind": "team_or_shared_result_as_personal"})
if problems:
for problem in problems:
problem["line"] = sentence.line
problem["evidence_ids"] = list(sentence.evidence_ids)
problem["text"] = _redact_context(sentence.text, blocked_terms)
return problems
def check_draft_text(
text: str,
*,
file_name: str = "<text>",
data_dir: Path | None = None,
limit: int | None = None,
minimum: int | None = None,
count_spaces: bool | None = None,
required_format: str | None = None,
source_path: Path | None = None,
) -> dict[str, Any]:
if data_dir is not None:
validation = validate_workspace(data_dir)
if validation["status"] == "BLOCK":
messages = "; ".join(
item["message"]
for item in validation["issues"]
if item["status"] == "BLOCK"
)
raise HarnessError(f"워크스페이스 검증이 필요합니다: {messages}")
comments, malformed = _parse_comments(text)
sentences = _sentences(text, comments)
draft_binding = (
_draft_binding(data_dir, source_path)
if data_dir is not None and source_path is not None
else None
)
catalog, catalog_available, catalog_error = _evidence_catalog(data_dir)
blocked_terms = _privacy_terms(data_dir)
(
inferred_limit,
inferred_minimum,
inferred_count_spaces,
inferred_format,
) = _inferred_constraints(data_dir, source_path)
effective_limit = limit if limit is not None else inferred_limit
effective_minimum = minimum if minimum is not None else inferred_minimum
effective_count_spaces = inferred_count_spaces if count_spaces is None else count_spaces
effective_format = required_format if required_format is not None else inferred_format
checks: list[dict[str, Any]] = []
unsafe_unicode: list[dict[str, Any]] = []
if text != unicodedata.normalize("NFC", text):
unsafe_unicode.append({"reason": "not_nfc_normalized"})
for index, character in enumerate(text):
category = unicodedata.category(character)
if category == "Cf" or (category == "Cc" and character not in "\n\r\t"):
unsafe_unicode.append(
{
"line": text.count("\n", 0, index) + 1,
"reason": "invisible_or_control_character",
"codepoint": f"U+{ord(character):04X}",
}
)
if unsafe_unicode:
checks.append(
_check(
"BLOCK",
"UNSAFE_UNICODE",
"정규화되지 않은 한글이나 보이지 않는 제어 문자를 사용할 수 없습니다.",
unsafe_unicode,
)
)
else:
checks.append(_check("PASS", "UNSAFE_UNICODE", "본문 Unicode 형식이 안전합니다."))
malformed_ids = [
{"line": item["line"], "reason": "invalid evidence id"}
for item in comments
for evidence_id in item["ids"]
if not VALID_EVIDENCE_ID.fullmatch(evidence_id)
]
empty_comments = [item["line"] for item in comments if not item["ids"]]
unexpected_comments = [
{
"line": text.count("\n", 0, match.start()) + 1,
"reason": "unexpected HTML comment",
}
for match in ANY_HTML_COMMENT.finditer(text)
if VALID_COMMENT.fullmatch(match.group(0)) is None
]
if malformed or malformed_ids or empty_comments or unexpected_comments:
details: list[Any] = []
details.extend(
{"line": item["line"], "reason": "malformed evidence comment"}
for item in malformed
)
details.extend(malformed_ids)
details.extend({"line": line, "reason": "empty evidence list"} for line in empty_comments)
details.extend(unexpected_comments)
checks.append(
_check(
"BLOCK",
"EVIDENCE_SYNTAX",
"근거 주석은 <!-- evidence: E001,V001 --> 형식이어야 합니다.",
details,
)
)
else:
checks.append(_check("PASS", "EVIDENCE_SYNTAX", "근거 주석 형식이 유효합니다."))
placeholders = [
{
"line": text.count("\n", 0, match.start()) + 1,
"reason": "unfinished placeholder or work note",
}
for match in PLACEHOLDER_PATTERN.finditer(VALID_COMMENT.sub("", text))
]
if placeholders:
checks.append(
_check(
"BLOCK",
"UNFINISHED_PLACEHOLDER",
"[확인 필요], TODO 같은 미완성 표시를 제출용 초안에 남길 수 없습니다.",
placeholders,
)
)
else:
checks.append(_check("PASS", "UNFINISHED_PLACEHOLDER", "미완성 자리표시자가 없습니다."))
referenced_ids = tuple(
dict.fromkeys(
evidence_id
for item in comments
for evidence_id in item["ids"]
if VALID_EVIDENCE_ID.fullmatch(evidence_id)
)
)
evidence_problems: list[dict[str, Any]] = []
if referenced_ids and not catalog_available:
evidence_problems.append({"reason": catalog_error or "근거 카탈로그를 읽을 수 없습니다."})
elif catalog_available:
for evidence_id in referenced_ids:
record = catalog.get(evidence_id)
if record is None:
evidence_problems.append({"id": evidence_id, "reason": "unknown"})
elif not record.get("verified"):
evidence_problems.append({"id": evidence_id, "reason": "unverified"})
elif not record.get("allowed"):
evidence_problems.append({"id": evidence_id, "reason": "private_or_not_allowed"})
if evidence_problems:
checks.append(
_check(
"BLOCK",
"EVIDENCE_ELIGIBILITY",
"없거나 미확인·비공개인 근거를 사용할 수 없습니다.",
evidence_problems,
)
)
else:
checks.append(
_check("PASS", "EVIDENCE_ELIGIBILITY", "참조한 근거가 모두 확인되고 사용 허용되었습니다.")
)
if draft_binding is not None:
approved_evidence = draft_binding[1].get("outline", {}).get("evidence_ids", [])
approved_set = {
evidence_id for evidence_id in approved_evidence if isinstance(evidence_id, str)
}
outside_outline = [
{"id": evidence_id, "reason": "not_in_approved_outline"}
for evidence_id in referenced_ids
if evidence_id not in approved_set
]
if outside_outline:
checks.append(
_check(
"BLOCK",
"OUTLINE_EVIDENCE_SCOPE",
"승인 개요에 없는 근거는 이 문항 초안에 사용할 수 없습니다.",
outside_outline,
)
)
else:
checks.append(
_check(
"PASS",
"OUTLINE_EVIDENCE_SCOPE",
"모든 근거가 승인 개요의 범위 안에 있습니다.",
)
)
composed_story_lines: list[dict[str, Any]] = []
if catalog_available:
for sentence in sentences:
story_roots = sorted(
{
catalog[evidence_id].get("story_id")
for evidence_id in sentence.evidence_ids
if evidence_id in catalog
and isinstance(catalog[evidence_id].get("story_id"), str)
}
)
if len(story_roots) > 1:
composed_story_lines.append(
{"line": sentence.line, "story_ids": story_roots}
)
if composed_story_lines:
checks.append(
_check(
"BLOCK",
"MULTI_STORY_COMPOSITION",
"서로 다른 경험을 한 문장·사건처럼 합칠 수 없습니다. 경험별 문단을 분리하세요.",
composed_story_lines,
)
)
else:
checks.append(
_check(
"PASS",
"MULTI_STORY_COMPOSITION",
"한 문장 안에서 서로 다른 경험을 합성하지 않았습니다.",
)
)
detected_factual_sentences = [
sentence
for sentence in sentences
if NUMBER_PATTERN.search(sentence.text)
or _korean_number_claims(sentence.text)
or FACT_PATTERN.search(sentence.text)
]
factual_sentences = sentences if draft_binding is not None else detected_factual_sentences
unsupported_facts = [
{
"line": sentence.line,
"text": _redact_context(sentence.text, blocked_terms),
}
for sentence in factual_sentences
if not sentence.evidence_ids
]
if unsupported_facts:
checks.append(
_check(
"BLOCK",
"FACT_EVIDENCE_COVERAGE",
(
"문항에 연결된 초안의 각 본문 문단에는 근거 주석이 필요합니다."
if draft_binding is not None
else "경험·성과로 보이는 문장에 근거 주석이 없습니다."
),
unsupported_facts,
)
)
elif factual_sentences:
checks.append(_check("PASS", "FACT_EVIDENCE_COVERAGE", "감지된 사실 문장에 근거가 연결되었습니다."))
else:
checks.append(_check("PASS", "FACT_EVIDENCE_COVERAGE", "근거가 필요한 사실 문장이 감지되지 않았습니다."))
if draft_binding is not None and not sentences:
checks.append(_check("BLOCK", "EMPTY_DRAFT", "확정할 본문이 비어 있습니다."))
grounding_problems = [
problem
for sentence in factual_sentences
if sentence.evidence_ids
for problem in _grounding_problems(sentence, catalog, blocked_terms)
]
if grounding_problems:
checks.append(
_check(
"BLOCK",
"CLAIM_GROUNDING",
"초안 표현이 연결 근거에서 확인되지 않습니다. 근거와 같은 구체 명사를 "
"하나 이상 남기거나, 본인 역할·회사 사실을 다시 확인하세요.",
grounding_problems,
)
)
else:
checks.append(
_check(
"PASS",
"CLAIM_GROUNDING",
"감지된 기술·행동 표현이 연결 근거 범위 안에 있습니다.",
)
)
unsupported_numbers: list[dict[str, Any]] = []
for sentence in sentences:
number_claims = list(NUMBER_CLAIM_PATTERN.finditer(sentence.text))
korean_number_claims = _korean_number_claims(sentence.text)
if not number_claims and not korean_number_claims:
continue
if not sentence.evidence_ids:
unsupported_numbers.append(
{
"line": sentence.line,
"text": _redact_context(sentence.text, blocked_terms),
"reason": "no_annotation",
}
)
continue
if not catalog_available:
continue
records = [catalog[item] for item in sentence.evidence_ids if item in catalog]
for claim in number_claims:
token = claim.group("number")
unit = claim.group("unit")
if not _number_claim_supported(token, unit, records):
unsupported_numbers.append(
{
"line": sentence.line,
"text": _redact_context(sentence.text, blocked_terms),
"number": (
"[민감정보 제거]"
if _contains_sensitive_text(sentence.text, blocked_terms)
else token
),
"reason": "number_not_found_in_linked_evidence",
}
)
for claim in korean_number_claims:
token = claim.group("number")
unit = claim.group("unit")
if not _korean_number_claim_supported(token, unit, records):
unsupported_numbers.append(
{
"line": sentence.line,
"text": _redact_context(sentence.text, blocked_terms),
"number": "[근거에 없는 한글 수량]",
"reason": "number_not_found_in_linked_evidence",
}
)
if unsupported_numbers:
checks.append(
_check(
"BLOCK",
"UNSUPPORTED_NUMBER",
"숫자가 연결된 확인 근거에 존재하지 않습니다.",
unsupported_numbers,
)
)
else:
checks.append(_check("PASS", "UNSUPPORTED_NUMBER", "근거 없는 숫자 주장이 없습니다."))
visible_text = VALID_COMMENT.sub("", text)
inspection_text = text
sensitive_hits: list[dict[str, Any]] = []
for label, pattern in SENSITIVE_PATTERNS.items():
for match in pattern.finditer(inspection_text):
sensitive_hits.append(
{"line": inspection_text.count("\n", 0, match.start()) + 1, "kind": label}
)
for term in blocked_terms:
start = 0
while True:
position = inspection_text.find(term, start)
if position < 0:
break
sensitive_hits.append(
{
"line": inspection_text.count("\n", 0, position) + 1,
"kind": "profile.json privacy.do_not_use와 일치",
}
)
start = position + len(term)
if sensitive_hits:
checks.append(
_check(
"BLOCK",
"SENSITIVE_DATA",
"개인정보·비밀값 또는 사용 금지 문자열이 감지되었습니다.",
sensitive_hits,
)
)
else:
checks.append(_check("PASS", "SENSITIVE_DATA", "등록된 민감정보 패턴이 없습니다."))
private_url_lines: set[int] = set()
public_url_lines: set[int] = set()
for match in URL_PATTERN.finditer(visible_text):
line = visible_text.count("\n", 0, match.start()) + 1
(private_url_lines if _is_private_url(match.group(0)) else public_url_lines).add(line)
for match in BARE_ENDPOINT_PATTERN.finditer(visible_text):
if _is_private_bare_endpoint(match.group(0)):
private_url_lines.add(visible_text.count("\n", 0, match.start()) + 1)
if private_url_lines:
checks.append(
_check(
"BLOCK",
"URL_REVIEW",
"내부·사설 URL로 보이는 값이 있습니다.",
({"line": line} for line in sorted(private_url_lines)),
)
)
elif public_url_lines:
checks.append(
_check(
"WARN",
"URL_REVIEW",
"URL이 포함되어 있습니다. 공개 가능한 공식 링크인지 확인하세요.",
({"line": line} for line in sorted(public_url_lines)),
)
)
else:
checks.append(_check("PASS", "URL_REVIEW", "본문에 URL이 없습니다."))
cliche_hits = [phrase for phrase in CLICHES if phrase in visible_text]
if cliche_hits:
checks.append(
_check("WARN", "CLICHE", "상투적인 표현을 구체적인 행동 언어로 바꾸는 편이 좋습니다.", cliche_hits)
)
else:
checks.append(_check("PASS", "CLICHE", "등록된 상투 표현이 감지되지 않았습니다."))
long_sentences = [
{
"line": sentence.line,
"length": len(sentence.text),
"text": _redact_context(sentence.text, blocked_terms),
}
for sentence in sentences
if len(sentence.text) > 100
]
if long_sentences:
checks.append(
_check("WARN", "LONG_SENTENCE", "100자를 넘는 문장을 나누어 읽기 쉽게 만드세요.", long_sentences)
)
else:
checks.append(_check("PASS", "LONG_SENTENCE", "지나치게 긴 문장이 없습니다."))
endings: list[tuple[str, int]] = []
for sentence in sentences:
match = ENDING_PATTERN.search(sentence.text)
if match:
endings.append((match.group(1), sentence.line))
repeated: list[dict[str, Any]] = []
run_ending: str | None = None
run_lines: list[int] = []
for ending, line in endings + [("", -1)]:
if ending == run_ending:
run_lines.append(line)
continue
if run_ending and len(run_lines) >= 3:
repeated.append({"ending": run_ending, "lines": run_lines})
run_ending = ending
run_lines = [line] if ending else []
if repeated:
checks.append(
_check("WARN", "REPEATED_ENDING", "같은 종결 표현이 3문장 이상 연속됩니다.", repeated)
)
else:
checks.append(_check("PASS", "REPEATED_ENDING", "종결 표현의 연속 반복이 없습니다."))
format_details: list[dict[str, Any]] = []
if effective_format == "plain_text":
plain_body = VALID_COMMENT.sub("", text)
markdown_pattern = re.compile(r"(?m)^\s*(?:#{1,6}\s|[-*+]\s|>\s|```|\|.*\|\s*$)")
format_details = [
{"line": plain_body.count("\n", 0, match.start()) + 1}
for match in markdown_pattern.finditer(plain_body)
]
if format_details:
checks.append(
_check(
"BLOCK",
"REQUIRED_FORMAT",
"plain_text 문항에 Markdown 구조가 포함되어 있습니다.",
format_details,
)
)
else:
checks.append(_check("PASS", "REQUIRED_FORMAT", "요구된 본문 형식을 지켰습니다."))
character_count = _visible_character_count(text, effective_count_spaces)
if effective_limit is not None and character_count > effective_limit:
checks.append(
_check(
"BLOCK",
"LENGTH",
f"분량 {character_count}자가 제한 {effective_limit}자를 초과합니다.",
)
)
elif effective_minimum is not None and character_count < effective_minimum:
checks.append(
_check(
"BLOCK",
"LENGTH",
f"분량 {character_count}자가 최소 {effective_minimum}자보다 짧습니다.",
)
)
elif effective_limit is None:
checks.append(_check("PASS", "LENGTH", f"현재 분량은 {character_count}자이며 제한이 설정되지 않았습니다."))
elif character_count < max(1, int(effective_limit * 0.60)):
checks.append(
_check(
"WARN",
"LENGTH",
f"분량 {character_count}자가 제한 {effective_limit}자의 60%보다 짧습니다.",
)
)
elif character_count > int(effective_limit * 0.90):
checks.append(
_check(
"WARN",
"LENGTH",
f"분량 {character_count}자가 제한 {effective_limit}자의 90%를 넘어 수정 여유가 작습니다.",
)
)
else:
checks.append(_check("PASS", "LENGTH", f"분량 {character_count}자가 제한 {effective_limit}자 안입니다."))
overall = max((item["status"] for item in checks), key=SEVERITY.get, default="PASS")
counts = {
level.lower(): sum(item["status"] == level for item in checks)
for level in ("PASS", "WARN", "BLOCK")
}
return {
"status": overall,
"file": file_name,
"character_count": character_count,
"character_limit": effective_limit,
"character_minimum": effective_minimum,
"count_spaces": effective_count_spaces,
"required_format": effective_format,
"draft_sha256": hashlib.sha256(text.encode("utf-8")).hexdigest(),
"evidence_ids": list(referenced_ids),
"summary": counts,
"checks": checks,
}
def check_draft_file(
path: Path,
*,
workspace: str | Path | None = None,
limit: int | None = None,
minimum: int | None = None,
count_spaces: bool | None = None,
required_format: str | None = None,
) -> dict[str, Any]:
path = _resolve_path(path, "초안 경로")
try:
text = path.read_text(encoding="utf-8")
except FileNotFoundError as exc:
raise HarnessError(f"초안 파일이 없습니다: {path}") from exc
except OSError as exc:
raise HarnessError(f"초안 파일을 읽을 수 없습니다: {path} ({exc})") from exc
except UnicodeError as exc:
raise HarnessError(f"초안 파일은 UTF-8이어야 합니다: {path}") from exc
if workspace is not None:
data_dir = _data_directory(workspace)
else:
data_dir = _discover_data_directory(path) or _discover_data_directory(Path.cwd())
return check_draft_text(
text,
file_name=str(path),
data_dir=data_dir,
limit=limit,
minimum=minimum,
count_spaces=count_spaces,
required_format=required_format,
source_path=path,
)
def _print_validation(report: Mapping[str, Any]) -> None:
print(f"{report['status']} {report['workspace']}")
for item in report["issues"]:
print(f"[{item['status']}] {item['code']}: {item['message']}")
def _print_draft_report(report: Mapping[str, Any]) -> None:
limit = report.get("character_limit")
length = f"{report['character_count']}자" + (f"/{limit}자" if limit else "")
print(f"{report['status']} {report['file']} ({length})")
for item in report["checks"]:
print(f"[{item['status']}] {item['code']}: {item['message']}")
for detail in item.get("details", []):
print(f" - {json.dumps(detail, ensure_ascii=False) if isinstance(detail, (dict, list)) else detail}")
def _command_init(args: argparse.Namespace) -> int:
report = initialize_workspace(args.workspace)
print(f"PASS {report['workspace']}")
if report["created"]:
print("created: " + ", ".join(report["created"]))
if report["skipped"]:
print("kept existing: " + ", ".join(report["skipped"]))
return 0
def _command_status(args: argparse.Namespace) -> int:
data_dir = _data_directory(args.workspace)
report = workspace_status(data_dir)
if args.json:
_json_print(report)
else:
print(f"{report['status']} 단계={report['stage']} 작업공간={report['workspace']}")
print("작성 컨텍스트 준비=" + str(report["ready_for_context"]).lower())
print("내보내기 준비=" + str(report["ready_for_export"]).lower())
active_application = report.get("active_application_id") or "없음"
active_question = report.get("active_question_id") or "없음"
print(f"활성 지원처/문항={active_application}/{active_question}")
counts = report.get("counts", {})
if counts:
print(
"경험="
f"{counts.get('stories', 0)}개 "
f"(확인·허용 {counts.get('eligible_stories', 0)}개), "
f"지원처={counts.get('applications', 0)}개"
)
print("다음: " + NEXT_MESSAGES.get(report["stage"], "validate 결과를 확인하세요."))
for command in _next_exact_commands(data_dir, report):
print("실행: " + command)
for item in report.get("issues", []):
print(f"[{item['status']}] {item['code']}: {item['message']}")
return 1 if report["status"] == "BLOCK" else 0
def _command_next(args: argparse.Namespace) -> int:
data_dir = _data_directory(args.workspace)
report = workspace_status(data_dir)
print(f"다음 [{report['stage']}] {NEXT_MESSAGES.get(report['stage'], 'status를 확인하세요.')}")
for command in _next_exact_commands(data_dir, report):
print("실행: " + command)
return 1 if report["status"] == "BLOCK" else 0
def _command_validate(args: argparse.Namespace) -> int:
report = validate_workspace(_data_directory(args.workspace))
_print_validation(report)
return 1 if report["status"] == "BLOCK" else 0
def _command_context(args: argparse.Namespace) -> int:
context = build_context(_data_directory(args.workspace), args.application, args.question)
_json_print(context)
return 0
def _positive_int(value: str) -> int:
try:
parsed = int(value)
except ValueError as exc:
raise argparse.ArgumentTypeError("양의 정수여야 합니다.") from exc
if parsed <= 0:
raise argparse.ArgumentTypeError("양의 정수여야 합니다.")
return parsed
def _nonnegative_int(value: str) -> int:
try:
parsed = int(value)
except ValueError as exc:
raise argparse.ArgumentTypeError("0 이상의 정수여야 합니다.") from exc
if parsed < 0:
raise argparse.ArgumentTypeError("0 이상의 정수여야 합니다.")
return parsed
def _command_check_draft(args: argparse.Namespace) -> int:
report = check_draft_file(
_resolve_path(args.file, "초안 경로"),
workspace=args.workspace,
limit=args.limit,
minimum=args.minimum,
count_spaces=args.count_spaces,
required_format=args.required_format,
)
if args.json:
_json_print(report)
else:
_print_draft_report(report)
return 1 if report["status"] == "BLOCK" else 0
def _command_approve(args: argparse.Namespace) -> int:
if not args.confirm_all:
raise HarnessError(
"최종본의 사실·본인 역할·개인정보·문항 적합성·내 말투·면접 설명 가능성을 "
"직접 확인한 뒤 --confirm-all을 붙이세요."
)
source = _resolve_path(args.file, "초안 경로")
data_dir = _discover_data_directory(source)
if data_dir is None:
raise HarnessError("초안의 .cover-letter 워크스페이스를 찾을 수 없습니다.")
validation = validate_workspace(data_dir)
if validation["status"] == "BLOCK":
messages = "; ".join(
item["message"] for item in validation["issues"] if item["status"] == "BLOCK"
)
raise HarnessError(f"워크스페이스 검증이 필요합니다: {messages}")
binding = _draft_binding(data_dir, source, required=True)
if binding is None:
raise HarnessError("초안과 문항의 연결을 확인할 수 없습니다.")
application, question = binding
documents = _load_workspace(data_dir)
if documents["profile.json"].get("confirmed") is not True:
raise HarnessError("확인되지 않은 profile의 초안은 확정할 수 없습니다.")
if application.get("confirmed") is not True:
raise HarnessError("확인되지 않은 application의 초안은 확정할 수 없습니다.")
if question.get("outline_approved") is not True:
raise HarnessError("승인되지 않은 개요의 초안은 확정할 수 없습니다.")
try:
text = source.read_text(encoding="utf-8")
except FileNotFoundError as exc:
raise HarnessError(f"초안 파일이 없습니다: {source}") from exc
except OSError as exc:
raise HarnessError(f"초안 파일을 읽을 수 없습니다: {source} ({exc})") from exc
except UnicodeError as exc:
raise HarnessError(f"초안 파일은 UTF-8이어야 합니다: {source}") from exc
limit, minimum, count_spaces, required_format = _question_constraints(question)
report = check_draft_text(
text,
file_name=str(source),
data_dir=data_dir,
limit=limit,
minimum=minimum,
count_spaces=count_spaces,
required_format=required_format,
source_path=source,
)
if report["status"] == "BLOCK":
blocking = "; ".join(
item["message"] for item in report["checks"] if item["status"] == "BLOCK"
)
raise HarnessError(f"BLOCK 초안은 확정할 수 없습니다: {blocking}")
application_id = str(application.get("id"))
question_id = str(question.get("id"))
context = build_context(data_dir, application_id, question_id)
applications_document = documents["applications.json"]
persisted_application = next(
item
for item in applications_document["applications"]
if _is_object(item) and item.get("id") == application_id
)
persisted_question = next(
item
for item in persisted_application["questions"]
if _is_object(item) and item.get("id") == question_id
)
persisted_question["status"] = "user_approved"
persisted_question["verification"] = {
**{flag: True for flag in REQUIRED_VERIFICATION_FLAGS},
"draft_sha256": report["draft_sha256"],
"context_sha256": _json_digest(context),
}
_write_json_private_atomic(data_dir / "applications.json", applications_document)
session = documents["session.json"]
session["stage"] = "USER_APPROVED"
session["active_application_id"] = application_id
session["active_question_id"] = question_id
_write_json_private_atomic(data_dir / "session.json", session)
post_validation = validate_workspace(data_dir)
if post_validation["status"] == "BLOCK":
raise HarnessError("확정 상태 저장 후 무결성 검증에 실패했습니다. validate를 실행하세요.")
warning_codes = [
item["code"] for item in report["checks"] if item["status"] == "WARN"
]
warning_suffix = f" warnings={','.join(warning_codes)}" if warning_codes else ""
print(
f"PASS approved {application_id}/{question_id} "
f"({report['character_count']}자){warning_suffix}"
)
return 0
def _command_export(args: argparse.Namespace) -> int:
source = _resolve_path(args.input, "입력 초안 경로")
destination = _resolve_path(args.output, "출력 경로")
if source == destination:
raise HarnessError("입력 초안과 출력 파일은 같은 경로일 수 없습니다.")
if source.exists() and destination.exists():
try:
if os.path.samefile(source, destination):
raise HarnessError("입력 초안과 출력 파일이 같은 파일을 가리킵니다.")
except OSError as exc:
raise HarnessError(f"입출력 파일을 확인할 수 없습니다: {exc}") from exc
if destination.exists() and not args.force:
raise HarnessError(f"출력 파일이 이미 있습니다: {destination} (--force로 덮어쓰기)")
data_dir = _discover_data_directory(source)
if data_dir is None:
raise HarnessError("초안의 .cover-letter 워크스페이스를 찾을 수 없습니다.")
validation = validate_workspace(data_dir)
if validation["status"] == "BLOCK":
messages = "; ".join(
item["message"] for item in validation["issues"] if item["status"] == "BLOCK"
)
raise HarnessError(f"워크스페이스 검증이 필요합니다: {messages}")
binding = _draft_binding(data_dir, source, required=True)
if binding is None:
raise HarnessError("초안과 문항의 연결을 확인할 수 없습니다.")
application, question = binding
documents = _load_workspace(data_dir)
profile = documents["profile.json"]
if profile.get("confirmed") is not True:
raise HarnessError("확인되지 않은 profile은 export할 수 없습니다.")
if application.get("confirmed") is not True:
raise HarnessError("확인되지 않은 application은 export할 수 없습니다.")
if question.get("outline_approved") is not True:
raise HarnessError("승인되지 않은 개요는 export할 수 없습니다.")
if question.get("status") != "user_approved":
raise HarnessError("question.status가 user_approved인 초안만 export할 수 있습니다.")
verification = question.get("verification", {})
if not _is_object(verification) or any(
verification.get(flag) is not True for flag in REQUIRED_VERIFICATION_FLAGS
):
raise HarnessError("사실·역할·개인정보·문항·문체·면접 확인을 모두 완료해야 합니다.")
context = build_context(data_dir, str(application.get("id")), str(question.get("id")))
if verification.get("context_sha256") != _json_digest(context):
raise HarnessError("사용자 승인 이후 사실·개요·작성 컨텍스트가 변경되었습니다. 다시 검증·확정하세요.")
try:
text = source.read_text(encoding="utf-8")
except FileNotFoundError as exc:
raise HarnessError(f"초안 파일이 없습니다: {source}") from exc
except OSError as exc:
raise HarnessError(f"초안 파일을 읽을 수 없습니다: {source} ({exc})") from exc
except UnicodeError as exc:
raise HarnessError(f"초안 파일은 UTF-8이어야 합니다: {source}") from exc
actual_digest = hashlib.sha256(text.encode("utf-8")).hexdigest()
if verification.get("draft_sha256") != actual_digest:
raise HarnessError("사용자 승인 이후 초안이 변경되었습니다. 다시 검증·확정하세요.")
limit, minimum, count_spaces, required_format = _question_constraints(question)
report = check_draft_text(
text,
file_name=str(source),
data_dir=data_dir,
limit=limit,
minimum=minimum,
count_spaces=count_spaces,
required_format=required_format,
source_path=source,
)
if report["status"] == "BLOCK":
blocking = "; ".join(
item["message"] for item in report["checks"] if item["status"] == "BLOCK"
)
raise HarnessError(f"BLOCK 초안은 export할 수 없습니다: {blocking}")
exported = _submission_text(text)
destination_parent_existed = destination.parent.exists()
destination.parent.mkdir(parents=True, mode=0o700, exist_ok=True)
if not destination_parent_existed:
destination.parent.chmod(0o700)
descriptor, temporary_name = tempfile.mkstemp(
prefix=".cover-letter-export-", dir=destination.parent
)
temporary_path = Path(temporary_name)
try:
os.fchmod(descriptor, 0o600)
with os.fdopen(descriptor, "w", encoding="utf-8") as handle:
handle.write(exported)
handle.flush()
os.fsync(handle.fileno())
if args.force:
os.replace(temporary_path, destination)
else:
try:
os.link(temporary_path, destination)
except FileExistsError as exc:
raise HarnessError(f"출력 파일이 이미 있습니다: {destination}") from exc
temporary_path.unlink()
finally:
if temporary_path.exists():
temporary_path.unlink()
count_with_spaces = _visible_character_count(exported, True)
count_without_spaces = _visible_character_count(exported, False)
print(
f"{report['status']} exported {destination} "
f"(공백 포함 {count_with_spaces}자, 공백 제외 {count_without_spaces}자)"
)
return 0
def build_parser() -> argparse.ArgumentParser:
parser = argparse.ArgumentParser(
prog="cover-letter",
description="근거와 사용 동의를 지키는 한국어 IT 자소서 로컬 하네스",
)
commands = parser.add_subparsers(dest="command", required=True)
init_parser = commands.add_parser("init", help=".cover-letter 워크스페이스 초기화")
init_parser.add_argument("workspace", nargs="?", default=".")
init_parser.set_defaults(handler=_command_init)
status_parser = commands.add_parser("status", help="현재 준비 단계 확인")
status_parser.add_argument("workspace", nargs="?", default=".")
status_parser.add_argument("--json", action="store_true", help="JSON으로 출력")
status_parser.set_defaults(handler=_command_status)
next_parser = commands.add_parser("next", help="다음 한 단계 안내")
next_parser.add_argument("workspace", nargs="?", default=".")
next_parser.set_defaults(handler=_command_next)
validate_parser = commands.add_parser("validate", help="JSON 스키마와 참조 검증")
validate_parser.add_argument("workspace", nargs="?", default=".")
validate_parser.set_defaults(handler=_command_validate)
context_parser = commands.add_parser("context", help="승인된 작성 컨텍스트 출력")
context_parser.add_argument("workspace", nargs="?", default=".")
context_parser.add_argument("--application", required=True, metavar="ID")
context_parser.add_argument("--question", required=True, metavar="ID")
context_parser.set_defaults(handler=_command_context)
check_parser = commands.add_parser("check-draft", help="근거·문체·분량 품질 게이트")
check_parser.add_argument("file")
check_parser.add_argument("--workspace", metavar="PATH")
check_parser.add_argument("--limit", type=_positive_int, metavar="N")
check_parser.add_argument("--minimum", type=_nonnegative_int, metavar="N")
count_group = check_parser.add_mutually_exclusive_group()
count_group.add_argument("--count-spaces", dest="count_spaces", action="store_true")
count_group.add_argument("--exclude-spaces", dest="count_spaces", action="store_false")
check_parser.set_defaults(count_spaces=None)
check_parser.add_argument(
"--format", dest="required_format", choices=("plain_text", "markdown")
)
check_parser.add_argument("--json", action="store_true", help="JSON으로 출력")
check_parser.set_defaults(handler=_command_check_draft)
approve_parser = commands.add_parser(
"approve", help="검증된 초안을 사용자가 최종 확인하고 해시로 잠금"
)
approve_parser.add_argument("file")
approve_parser.add_argument(
"--confirm-all",
action="store_true",
help="사실·역할·개인정보·문항·문체·면접 설명 가능성을 모두 직접 확인",
)
approve_parser.set_defaults(handler=_command_approve)
export_parser = commands.add_parser("export", help="검사를 통과한 초안에서 근거 주석 제거")
export_parser.add_argument("input")
export_parser.add_argument("output")
export_parser.add_argument("--force", action="store_true", help="기존 출력 파일 덮어쓰기")
export_parser.set_defaults(handler=_command_export)
return parser
def main(argv: Sequence[str] | None = None) -> int:
parser = build_parser()
args = parser.parse_args(argv)
try:
return int(args.handler(args))
except HarnessError as exc:
print(f"BLOCK: {exc}", file=sys.stderr)
return 1
except OSError as exc:
print(f"BLOCK: 파일 작업을 완료할 수 없습니다: {exc}", file=sys.stderr)
return 1
except (UnicodeError, RuntimeError) as exc:
print(f"BLOCK: 입력 파일이나 경로를 처리할 수 없습니다: {exc}", file=sys.stderr)
return 1
except KeyboardInterrupt:
print("중단되었습니다.", file=sys.stderr)
return 130
if __name__ == "__main__":
raise SystemExit(main())