Files
document-haness/build/lib/claridoc/providers/mock.py
T

288 lines
25 KiB
Python

from __future__ import annotations
import json
from typing import Any
from claridoc.models import Brief, Outline, SourcePack
from claridoc.providers.base import Provider, ProviderRequest, ProviderResponse
from claridoc.utils import extract_tag_json
class MockProvider(Provider):
"""Deterministic offline provider for contract and pipeline tests.
The mock deliberately avoids copying source excerpts into reader-facing prose. It
validates wiring and quality gates; it is not a substitute for a writing model.
"""
def generate(self, request: ProviderRequest) -> ProviderResponse:
if request.stage == "plan":
text = json.dumps(
extract_tag_json(request.prompt, "BASE_OUTLINE_JSON"),
ensure_ascii=False,
indent=2,
)
elif request.stage in {"draft", "revise"}:
brief = Brief.from_dict(extract_tag_json(request.prompt, "BRIEF_JSON"))
outline = Outline.from_dict(extract_tag_json(request.prompt, "OUTLINE_JSON"))
sources = SourcePack.from_dict(extract_tag_json(request.prompt, "SOURCE_PACK_JSON"))
text = _make_document(brief, outline, sources)
elif request.stage == "review":
lint = extract_tag_json(request.prompt, "DETERMINISTIC_LINT_JSON")
role = str(request.metadata.get("role", "logic"))
text = json.dumps(_make_review(lint, role), ensure_ascii=False, indent=2)
else:
text = "Mock provider received an unsupported stage."
return ProviderResponse(text=text, provider=self.name, model="deterministic-mock")
def check(self) -> dict[str, Any]:
return {
"provider": self.name,
"available": True,
"mode": "deterministic offline fixture",
"note": "Does not call an external model and does not measure prose quality.",
}
def _make_review(lint: dict[str, Any], role: str) -> dict[str, Any]:
raw_issues = lint.get("issues", [])
material = [item for item in raw_issues if item.get("severity") in {"blocker", "error"}]
score = max(55.0, min(96.0, float(lint.get("score", 80)) + (3 if not material else -3)))
dimensions = {
"reader_goal_alignment": score,
"information_architecture": score,
"logical_flow": score,
"decision_rationale": score,
"source_usefulness": score,
"reader_facing_prose": score,
"cognitive_load": min(100, score + 1),
"evidence_traceability": score,
"example_verifiability": score,
"scannability": min(100, score + 1),
"operational_safety": score,
"completeness_and_limits": score,
}
issues = [
{
"section": item.get("section")
or (f"line {item.get('line')}" if item.get("line") else "document"),
"problem": item.get("message", "deterministic finding"),
"why_it_matters": "It can interrupt the reader path or violate the document contract.",
"fix": item.get("suggestion") or "Resolve the deterministic finding directly.",
"severity": item.get("severity", "error"),
}
for item in material
]
return {
"score": score,
"dimension_scores": dimensions,
"issues": issues,
"strengths": [
f"The deterministic {role} fixture found the document contract inspectable."
],
"questions": [],
}
def _make_document(brief: Brief, outline: Outline, sources: SourcePack) -> str:
# `sources` is intentionally not rendered. Source IDs, paths, and access dates belong
# in provenance.md/evidence-map.json, which the pipeline creates separately.
_ = sources
lines: list[str] = [f"# {brief.title}", ""]
for section in outline.sections:
lines.extend([f"## {section.title}", ""])
body = (
_korean_body(brief, section.intent)
if brief.is_korean
else _english_body(brief, section.intent)
)
lines.extend(body)
lines.append("")
return "\n".join(lines).strip() + "\n"
def _korean_body(brief: Brief, intent: str) -> list[str]:
topics = ", ".join(brief.required_topics) or "핵심 구성요소"
scope = ", ".join(brief.scope)
non_scope = ", ".join(brief.non_scope) or "별도 비범위 없음"
prereq = ", ".join(brief.prerequisites) or "별도 선행 조건 없음"
technical_blog: dict[str, list[str]] = {
"problem_scene": [
f"작은 구현 선택처럼 보였던 문제가 실제 흐름을 따라가자 여러 경계에 걸쳐 있었다. {topics} 가운데 하나만 고치면 다른 지점에서 부하, 중복, 조립 비용, 복구 비용이 커질 수 있었다. 이 글은 다음 질문을 다룬다. **{brief.reader_goal}**",
f"핵심 판단은 명확하다. **{brief.core_message}** 여기서는 {scope}에 집중하며, {non_scope}까지 보편적인 결론으로 확대하지 않는다.",
],
"constraints": [
f"{topics}는 입력과 상태, 실패와 복구를 통해 서로 연결된다. 한 부분의 편의를 높이면 다른 경계로 부하나 중복, 복구 비용이 이동할 수 있어서 각 요소를 독립적으로 바꾸기 어려웠다.",
"근거의 역할도 서로 달랐다. 현재 구현, 결정 기록, 공식 동작, 다른 회사의 사례는 같은 단어를 사용하더라도 같은 사실을 증명하지 않는다. 프로젝트의 선택 이유는 그 이유를 직접 기록한 자료가 있을 때만 설명할 수 있다.",
],
"options": [
"검토할 선택지는 최소 두 가지다. 첫째, 현재 방식을 유지하고 문제가 드러난 지점만 보완한다. 변경 범위는 작지만 상호작용을 놓치기 쉽다. 둘째, 관련 요소를 하나의 정책 경계로 묶는다. 초기 설계와 검증 비용은 늘지만 판단 기준과 실패 범위를 함께 관리할 수 있다.",
"비교 기준은 구현량이 아니라 실패 시 부하가 어디로 이동하는지, 중복 부작용을 막을 수 있는지, 검증 결과를 관측할 수 있는지, 잘못됐을 때 되돌릴 수 있는지다. 실패한 시도나 제외한 대안도 같은 기준으로 설명해야 독자가 선택을 재현할 수 있다.",
],
"decision_rationale": [
f"이 글이 선택한 방향은 **{brief.core_message}** 여러 설정을 함께 다루기로 한 이유는 각각의 값이 서로의 안전 조건을 바꾸기 때문이다. 한 항목만 최적화하면 전체 요청 경로나 모듈 경계에서 예상하지 못한 비용이 발생한다.",
"대안은 설정을 완전히 분리하거나 편의를 위해 관련 경계를 넓게 허용하는 방식이다. 전자는 상호작용을 운영자에게 떠넘기고, 후자는 정책이 코어 안으로 번질 위험을 키운다. 따라서 초기 설계와 테스트 비용을 수용하되, 허용 범위와 금지 범위를 자동 검사하는 가드레일을 함께 둔다.",
],
"mechanism": [
"결정은 입력에서 관측까지 끊기지 않는 흐름으로 반영한다. 요청이나 변경이 들어오면 사전 조건을 확인하고, 같은 기준에서 실행 경로와 상태 변경 범위를 정한다. 실행 뒤에는 결과와 실패 신호를 기록해 성공, 중단, 복구 중 하나를 결정한다.",
"```text\n입력과 현재 상태\n → 안전 조건 확인\n → 한정된 실행 경로 선택\n → 상태 변경 또는 호출\n → 로그·지표·테스트 결과 관측\n → 확정 / 중단 / 복구\n```",
"이 흐름의 불변조건은 실패한 작업이 성공으로 기록되지 않고, 같은 입력을 다시 처리했을 때 허용하지 않은 부작용이 늘어나지 않는 것이다. 실제 글에서는 일반 명칭 대신 프로젝트의 모듈, 인터페이스, 테스트 이름을 사용한다.",
],
"evidence_verification": [
"검증은 주장마다 관측 가능한 증거를 붙이는 방식으로 설계한다. 구조적 경계는 빌드 규칙이나 정적 분석으로, 런타임 동작은 단위·통합 테스트와 로그·지표로, 실패 복구는 의도된 오류 주입과 롤백 확인으로 검증한다.",
f"성공 기준은 독자가 다음 목표를 반복 가능한 결과로 확인할 수 있는지다. **{brief.reader_goal}** 반대로 운영 배포, 장기 부하, 특정 장애 조합을 검증하지 않았다면 그 범위는 명시적으로 남겨야 한다. 로컬 테스트 통과를 운영 검증으로 확대해 쓰지 않는다.",
],
"tradeoffs": [
"얻는 것은 판단 기준의 일관성, 실패 범위의 가시성, 자동 검증 가능성이다. 잃는 것은 초기 설계 시간과 정책을 유지하는 비용이다. 작은 실험이나 폐기 예정 코드에서는 이 구조가 과할 수 있지만, 반복 사용되거나 장애 시 비용이 큰 경로에서는 그 비용이 가드레일로 작동한다.",
"이 선택은 보편 법칙이 아니다. 성공 기준을 관측할 수 없거나 관련 요소의 소유권이 분리돼 있다면 더 작은 경계가 나을 수 있다. 남은 위험은 자동 검사가 잡지 못하는 런타임 우회와 문서·구현 간 시차이며, 코드 리뷰와 주기적인 근거 재검증으로 보완한다.",
],
"conclusion": [
f"결국 지키려던 것은 특정 도구가 아니라 판단 가능한 경계다. **{brief.core_message}** 자신의 환경에서는 ‘왜 이 선택이 필요한가’, ‘대안보다 어떤 비용을 덜어 주는가’, ‘그 대가를 어떤 테스트가 제한하는가’를 연속해서 답할 수 있어야 한다.",
],
}
if intent in technical_blog:
return technical_blog[intent]
procedural: dict[str, list[str]] = {
"outcome": [f"완성 결과는 **{brief.reader_goal}**이다. {brief.core_message}", f"대상 범위는 {scope}이며 {non_scope}는 다루지 않는다."],
"goal": [f"목표는 **{brief.reader_goal}**이다. {brief.core_message}", f"이 절차는 {scope}에 적용하고 {non_scope}에는 적용하지 않는다."],
"prerequisites": [f"시작 전에 {prereq}를 준비한다. 권한, 초기 상태, 복구점을 확인하지 못하면 실행하지 않는다."],
"route": ["전체 경로는 준비 → 최소 변경 → 중간 확인 → 최종 검증 순서다. 각 체크포인트를 통과하기 전에는 다음 단계로 이동하지 않는다."],
"guided_steps": [
"1. 현재 상태와 기대 결과를 기록한다.\n2. 한 번에 하나의 유효한 변경만 적용한다.\n3. 예상 결과와 실제 결과를 비교하고 다르면 중단한다.",
"```bash\nprintf '%s\\n' 'replace with a read-only verification command'\n```",
],
"procedure": [
"1. 현재 상태를 조회하고 복구점을 만든다.\n2. 목표에 필요한 최소 변경을 적용한다.\n3. 읽기 전용 확인 명령으로 결과를 검증한다.",
"```bash\nprintf '%s\\n' 'verify current state'\n```",
],
"checkpoint": ["중간 체크포인트에서는 입력, 변경 대상, 예상 출력이 모두 일치하는지 확인한다. 하나라도 다르면 마지막 정상 상태로 돌아간다."],
"verification": [f"같은 입력으로 검증을 반복한다. 성공 기준은 {brief.reader_goal}이 관측되고 범위 밖 상태가 바뀌지 않는 것이다."],
"rollback": ["중단 조건은 예상 범위 밖 변경, 검증 실패, 관측 불능이다. 쓰기를 멈추고 기록한 복구점을 복원한 뒤 읽기 전용 검사로 원복을 확인한다."],
"troubleshooting": ["1. 증상을 같은 입력으로 재현한다.\n2. 정상 기준과 다른 첫 관측을 찾는다.\n3. 확인된 원인에만 최소 조치를 적용하고 같은 검증을 반복한다."],
"next_steps": ["다음 단계는 현재 성공 기준을 실제 환경의 테스트와 관측값으로 치환하고, 하나의 경계 조건을 추가해 같은 구조가 유지되는지 확인하는 것이다."],
}
if intent in procedural:
return procedural[intent]
generic: dict[str, list[str]] = {
"question": [f"이 문서가 답하는 질문은 {brief.reader_goal}이다. 핵심 답은 **{brief.core_message}** 범위는 {scope}이며 {non_scope}는 제외한다."],
"familiar_anchor": [f"익숙한 흐름인 입력 → 판단 → 실행 → 관측에 {topics}를 배치하면 새 개념의 위치를 파악하기 쉽다. 같은 점은 단계별 책임이고, 다른 점은 실패가 다음 처리에 누적될 수 있다는 점이다."],
"mental_model": ["멘털 모델은 입력, 판단 기준, 상태 변화, 관측 결과의 네 요소다. 각 요소의 소유자와 불변조건을 분리하면 구현 세부사항이 바뀌어도 인과 관계를 추적할 수 있다."],
"mechanism": ["시작 조건을 확인한 뒤 명시된 기준으로 경로를 선택한다. 실행 결과는 상태와 관측값으로 남고, 그 값이 다음 행동을 결정한다."],
"example": ["```text\n입력 → 기준 확인 → 제한된 실행 → 결과 관측 → 다음 결정\n```", "예시의 목적은 각 단계에서 무엇을 알고 무엇을 확인해야 하는지 드러내는 것이다."],
"alternatives": ["대안은 단순성, 변경 위험, 관측성, 복구성이라는 같은 기준으로 비교한다. 선택의 장점만 나열하지 않고 적용하지 않을 조건도 함께 둔다."],
"limits": ["이 설명은 책임과 성공 기준을 관측할 수 있을 때 유효하다. 입력이나 소유권이 불명확하면 모델이 결정을 대신하지 못한다."],
"summary": [f"추천 방향은 **{brief.core_message}** 적용 범위는 {scope}이며 {non_scope}는 의도적으로 제외한다."],
"context": [f"현재 문제는 {topics}의 책임과 경계가 분리되어 있지 않아 변경 영향과 실패 위치를 추적하기 어렵다는 점이다."],
"goals_non_goals": [f"목표는 {brief.reader_goal}이다. 비목표는 {non_scope}이며, 성공은 반복 가능한 검증 결과로 판정한다."],
"constraints": [f"기능 요구는 {topics}의 핵심 흐름을 만족하는 것이다. 고정 제약은 현재 호환성과 안전한 실패, 관측 가능성, 복구 가능성이다."],
"options": ["대안은 현재 방식 보완과 경계 재설계다. 두 선택지를 단순성, 변경 위험, 관측성, 복구성으로 비교하고 제외 이유를 기록한다."],
"decision": [f"선택은 **{brief.core_message}**이다. 현재 제약에서 실패와 복구 경계를 함께 지키기 위해서다. 초기 설계 비용을 수용하는 대신 자동 검증 가드레일을 둔다."],
"failure_modes": ["주요 실패 모드는 입력 불일치, 부분 성공, 의존성 지연, 관측 누락이다. 각 실패에 중단 조건과 복구 경로를 둔다."],
"rollout": ["관측 가능한 작은 단위로 배포하고, 오류율이나 상태 불일치가 증가하면 이전 경로로 되돌린다."],
"observability": ["로그, 지표, 추적을 주장과 연결하고 변경 전 기준선과 비교한다. 정상, 실패, 롤백 경로를 모두 확인한다."],
"risks_open": ["남은 위험과 가정은 검증 방법, 소유자, 결정 기한과 함께 기록한다. 근거가 없는 가정은 열린 질문으로 남긴다."],
"syntax": ["```text\noperation(required_input, optional_input=default) -> result | error\n```", "필수 요소, 선택 요소, 생략 시 동작을 구분한다."],
"parameters": ["| 이름 | 타입 | 필수 | 기본값 | 제약 |\n|---|---|---:|---|---|\n| `required_input` | 프로젝트 타입 | 예 | 없음 | 사전 조건 충족 |"],
"behavior": ["정상 조건에서는 입력 검증 후 정의된 상태 전이만 수행하고 결과 또는 명시된 오류를 반환한다."],
"errors": ["| 오류 | 발생 조건 | 호출자 조치 |\n|---|---|---|\n| 입력 오류 | 사전 조건 불충족 | 입력 수정 |\n| 상태 충돌 | 현재 상태 불일치 | 상태 재조회 |"],
"examples": ["```text\nvalid input -> explicit result\ninvalid precondition -> documented error\n```"],
"related": ["관련 항목은 입력 타입, 반환 타입, 오류 정의, 관측 방법처럼 현재 경계와 직접 맞닿은 항목으로 제한한다."],
"symptom": ["동일 입력에서 반복되는 로그, 상태, 지표를 정상 기준과 비교해 증상을 재현한다."],
"impact": ["영향 범위는 사용자, 요청, 데이터, 의존 서비스 순서로 확인한다. 범위가 커지면 즉시 중단하고 에스컬레이션한다."],
"safety": ["진단 전에 증거를 보존하고 자동 변경을 중지하며 복구점을 확인한다."],
"diagnosis": ["1. 증상을 재현한다.\n2. 정상 기준과 다른 첫 관측을 찾는다.\n3. 입력, 상태, 의존성, 자원 경로로 분기한다."],
"causes": ["관측과 원인을 분리한다. 로그 한 줄만으로 확정하지 않고 반증 가능한 확인을 추가한다."],
"fixes": ["확인된 원인에만 최소 조치를 적용하고, 같은 진단으로 원인이 사라졌는지 확인한다."],
"prevention": ["같은 실패를 조기에 잡는 검사와 관측을 추가하고 소유자를 지정한다."],
"action": [f"실무에서는 {brief.reader_goal}을 관측 가능한 기준으로 바꾸고, 실패 조건과 복구 경로를 먼저 확인한다."],
"implications": ["구현 선택보다 입력, 상태 전이, 관측, 복구의 경계를 먼저 합의하면 세부 기술이 바뀌어도 판단 기준을 유지할 수 있다."],
}
return generic.get(intent, [f"**{brief.core_message}** {topics}를 입력, 판단, 상태 변화, 관측의 흐름으로 설명한다."])
def _english_body(brief: Brief, intent: str) -> list[str]:
topics = ", ".join(brief.required_topics) or "the key components"
scope = ", ".join(brief.scope)
non_scope = ", ".join(brief.non_scope) or "no declared non-scope"
prereq = ", ".join(brief.prerequisites) or "no additional prerequisite"
blog: dict[str, list[str]] = {
"problem_scene": [
f"A change that looked local became a boundary problem when the team followed state, failure, and recovery end to end. The practical question is how to {brief.reader_goal}. **{brief.core_message}**",
f"The discussion stays within {scope}. It does not claim that the same decision applies to {non_scope}.",
],
"constraints": [
f"The hard part is that {topics} do not move independently. A convenience at one boundary can shift load, duplication, or recovery cost to another boundary. Current implementation facts, decision history, official behavior, and external precedent must also be treated as different kinds of evidence.",
],
"options": [
"The first option is to preserve the current structure and patch only the visible failure. It limits change but can hide interactions. The second option is to define one policy boundary for the related decisions. It costs more up front but makes ownership, failure behavior, and verification explicit.",
"Both options should be compared on the same criteria: failure amplification, duplicate side effects, observability, reversibility, and maintenance cost. A rejected approach is useful only when the rejection condition is stated rather than implied.",
],
"decision_rationale": [
f"The selected direction is **{brief.core_message}** It was chosen because the related values change one another's safety conditions; optimizing one value in isolation can make the complete path less safe.",
"The realistic alternatives are fully independent settings or broad framework convenience. The former pushes coordination to operators, while the latter weakens the boundary. The design accepts additional configuration and test cost, with an automated guardrail that keeps the permission narrow.",
],
"mechanism": [
"The mechanism connects input to observation without a hidden jump. It checks preconditions, selects a bounded path, changes only the owned state, records the outcome, and then chooses acceptance, stop, or recovery.",
"```text\ninput and current state\n -> safety check\n -> bounded execution path\n -> state change\n -> observable result\n -> accept / stop / recover\n```",
"The invariant is that a failed operation is never recorded as successful and repeated input does not create an unbounded side effect.",
],
"evidence_verification": [
"Verification maps each claim to an observable check. Build rules or static analysis cover structural boundaries; unit and integration tests cover behavior; logs and metrics cover runtime effects; a failure exercise covers stop and recovery behavior.",
f"Success means the reader can {brief.reader_goal} using repeatable observations. A local test must not be described as production validation, and untested failure combinations remain explicit limits.",
],
"tradeoffs": [
"The design gains consistent decisions, visible failure boundaries, and automated checks. It spends more time on policy definition and maintenance. That cost may be excessive for disposable experiments, but it becomes a guardrail on paths that are reused or expensive to fail.",
"This is a project-local choice, not a universal rule. A smaller boundary may be better when ownership is split or success cannot be observed. Runtime bypasses and documentation drift remain risks that require review and periodic evidence refresh.",
],
"conclusion": [
f"The durable lesson is not a specific tool. **{brief.core_message}** A reader should be able to ask why the choice exists, which alternative it displaced, which cost it accepts, and which test keeps that cost bounded.",
],
}
if intent in blog:
return blog[intent]
if intent in {"guided_steps", "procedure", "diagnosis"}:
return [
f"Prerequisites: {prereq}.",
"1. Record the current state and expected outcome.\n2. Apply the smallest valid action.\n3. Compare the observed result with the success criterion and stop on mismatch.",
"```bash\nprintf '%s\\n' 'replace with a read-only verification command'\n```",
]
if intent in {"worked_example", "example", "examples"}:
return [
"```text\ninput -> explicit decision -> bounded change -> observation -> verified result\n```",
"The example exposes every transition instead of presenting only the final code.",
]
if intent in {"verification", "evidence_verification", "checkpoint", "observability"}:
return [
"Repeat the check with the same input, compare expected and observed state, and record acceptance, stop, and recovery criteria before the change is accepted."
]
if intent in {"rollback", "rollout", "failure_modes", "fixes", "safety", "prevention"}:
return [
"Stop on an unexpected state, preserve evidence, restore the recorded checkpoint, and verify recovery with a read-only check."
]
if intent == "parameters":
return ["| Name | Type | Required | Default | Constraints |\n|---|---|---:|---|---|\n| `required_input` | project-defined | yes | none | valid precondition |"]
if intent == "errors":
return ["| Error | Condition | Response |\n|---|---|---|\n| Invalid input | precondition fails | correct input |\n| State conflict | current state differs | reload and decide |"]
if intent == "prerequisites":
return [f"Before starting, confirm {prereq}, permissions, the initial state, and a recovery checkpoint."]
if intent == "rollback":
return ["Stop on an unexpected state, restore the recorded checkpoint, and verify recovery with a read-only check."]
if intent in {"options", "alternatives", "tradeoffs", "limits", "decision"}:
return [
"Compare at least two realistic options using the same constraints. State why the choice was made, which cost was accepted, and which guardrail prevents the decision from expanding beyond its intended boundary."
]
if intent in {"outcome", "goal", "question", "summary"}:
return [
f"The goal is to {brief.reader_goal}. **{brief.core_message}** The scope is {scope}; {non_scope} is excluded."
]
if intent in {"route", "checkpoint", "next_steps"}:
return ["Use the route prepare -> bounded action -> checkpoint -> final verification, and do not advance after a failed checkpoint."]
return [
f"**{brief.core_message}** Explain {topics} through explicit inputs, choices, state changes, observations, limits, and recovery behavior."
]