From c7270fb02ca02446a59ad75465abc3318ba9595d Mon Sep 17 00:00:00 2001 From: ToToKr Date: Tue, 21 Jul 2026 14:40:24 +0900 Subject: [PATCH 01/24] =?UTF-8?q?fix(f5):=20round-1=20review=20remediation?= =?UTF-8?q?=20=E2=80=94=20risk-cap=20max,=20strict=20RiskEvent,=20md/FHIR/?= =?UTF-8?q?PDF=20hardening,=20F5=20chain-wiring,=20KR=20font=20portability?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Remediates round-1 review blockers on the F5 clinical hand-off pipeline, preserving the original deterministic/ZERO-LLM design: - handoff_generator: _event_risk returns MAX of parseable severity signals (risk_level + ctrs via CTRS_TO_RISK) so a contradictory pair never resolves downward (issue #21); RiskEvent-or-dict normalization via _event_view. - schemas/handoff: new RiskEvent model — strict risk_level/ctrs_level validation at the API boundary (rejects null + unicode-confusable CTRS like '①'), extra keys preserved, max_length=100. - f5: A6->A8 leak guard NFKC+casefold normalized (case/fullwidth dodge closed); generated_at timezone-aware. - services/f5_report: markdown/HTML injection hardening (_md_inline/_md_cell/_md_block); PDF exporter isolation + newline-dense paragraph cap; FHIR null-primitive omission w/ dataAbsentReason, XHTML escaping, well-formed-div + null-value validation. - services/trend_plotter: bundled SHA-pinned Korean font assets are first-class candidates (portable across macOS/slim containers). - continuous_test: F5 runs from the post-ledger path in BOTH single- and multi-session chains (was bypassed/stale). - tests: adversarial coverage for all of the above + test_handoff_input_validation.py. --- .../ai-server/src/agents/handoff_generator.py | 48 +++-- apps/ai-server/src/continuous_test.py | 55 +++++- apps/ai-server/src/f5.py | 19 +- apps/ai-server/src/schemas/handoff.py | 61 +++++- apps/ai-server/src/services/f5_report.py | 113 +++++++++-- apps/ai-server/src/services/trend_plotter.py | 22 ++- .../tests/test_continuous_test_f4.py | 7 +- .../tests/test_continuous_test_f5.py | 177 ++++++++++++++++++ apps/ai-server/tests/test_f5.py | 44 +++++ apps/ai-server/tests/test_f5_fhir.py | 63 +++++++ apps/ai-server/tests/test_f5_report.py | 89 +++++++++ .../ai-server/tests/test_handoff_generator.py | 12 ++ .../tests/test_handoff_input_validation.py | 130 +++++++++++++ .../tests/test_orchestrator_handoff_input.py | 2 +- apps/ai-server/tests/test_trend_plotter.py | 43 +++++ 15 files changed, 827 insertions(+), 58 deletions(-) create mode 100644 apps/ai-server/tests/test_handoff_input_validation.py diff --git a/apps/ai-server/src/agents/handoff_generator.py b/apps/ai-server/src/agents/handoff_generator.py index 0036f81..1463c94 100644 --- a/apps/ai-server/src/agents/handoff_generator.py +++ b/apps/ai-server/src/agents/handoff_generator.py @@ -4,6 +4,7 @@ import logging import time +from collections.abc import Sequence from typing import Any from pydantic import BaseModel @@ -13,7 +14,7 @@ from src.prompts.loader import PromptLoader from src.routing.model_router import ModelRouter from src.schemas.common import CTRS_TO_RISK, CTRSLevel, EvidencePacket, EvidenceSource, RiskLevel -from src.schemas.handoff import HandoffInput, HandoffOutput, SlotData +from src.schemas.handoff import HandoffInput, HandoffOutput, RiskEvent, SlotData logger = logging.getLogger(__name__) @@ -59,7 +60,7 @@ def _build_user_content(inp: HandoffInput) -> str: if inp.risk_events: parts.append("\n## 위험 이벤트") for idx, evt in enumerate(inp.risk_events, 1): - parts.append(f"- [{idx}] {evt}") + parts.append(f"- [{idx}] {_event_view(evt) or evt}") # OCR documents if inp.ocr_documents: @@ -101,28 +102,47 @@ def _find_missing_slots(slots: SlotData) -> list[str]: _RISK_BY_VALUE: dict[str, RiskLevel] = {r.value: r for r in RiskLevel} +def _event_view(evt: object) -> dict[str, Any] | None: + """Normalize a risk event (validated RiskEvent or plain dict) to a dict.""" + if isinstance(evt, RiskEvent): + return evt.model_dump(exclude_none=True) + if isinstance(evt, dict): + return evt + return None + + def _event_risk(evt: object) -> RiskLevel: """Best-effort severity of a single risk event. - Prefers an explicit ``risk_level``; otherwise maps ``ctrs_level`` via - CTRS_TO_RISK; a present-but-unlabelled event keeps the "at least medium" - floor. + Returns the MAX of the parseable severity signals (explicit + ``risk_level`` and ``ctrs_level`` via CTRS_TO_RISK) — a contradictory + pair like risk_level="none" + ctrs_level="1" must never resolve + downward (issue #21: "return the max"). A present-but-unlabelled event + keeps the "at least medium" floor. CTRS parsing is ASCII-strict — + unicode digits like "١" never parse (route inputs are already rejected + at the schema boundary). """ - if not isinstance(evt, dict): + view = _event_view(evt) + if view is None: return RiskLevel.medium - raw = str(evt.get("risk_level", "")).strip().lower() + candidates: list[RiskLevel] = [] + raw = str(view.get("risk_level", "")).strip().lower() if raw in _RISK_BY_VALUE: - return _RISK_BY_VALUE[raw] - ctrs_raw = str(evt.get("ctrs_level", "")).strip() - if ctrs_raw.isdigit(): + candidates.append(_RISK_BY_VALUE[raw]) + ctrs_raw = str(view.get("ctrs_level", "")).strip() + if ctrs_raw.isascii() and ctrs_raw.isdigit(): try: - return CTRS_TO_RISK.get(CTRSLevel(int(ctrs_raw)), RiskLevel.medium) + ctrs_risk = CTRS_TO_RISK.get(CTRSLevel(int(ctrs_raw))) + if ctrs_risk is not None: + candidates.append(ctrs_risk) except ValueError: pass - return RiskLevel.medium + if not candidates: + return RiskLevel.medium + return max(candidates, key=lambda r: _RISK_ORDER[r]) -def _detect_risk_level(risk_events: list[dict[str, str]]) -> RiskLevel: +def _detect_risk_level(risk_events: Sequence[RiskEvent | dict[str, str]]) -> RiskLevel: """Return the maximum severity across all risk events (none if empty).""" if not risk_events: return RiskLevel.none @@ -356,7 +376,7 @@ def _next_id(prefix: str) -> str: evidence_id=_next_id("risk"), source_type=EvidenceSource.risk_event, source_ref="Safety Agent", - content_summary=str(evt)[:120], + content_summary=str(_event_view(evt) or evt)[:120], ) ) diff --git a/apps/ai-server/src/continuous_test.py b/apps/ai-server/src/continuous_test.py index 9f4b84b..1670792 100644 --- a/apps/ai-server/src/continuous_test.py +++ b/apps/ai-server/src/continuous_test.py @@ -950,6 +950,26 @@ async def run_multi_session_chain( if run_f4 and not halted: f4_result = await _run_f4_analysis(persona_id, out_dir) all_results.append(f4_result) + # F5 runs from the post-ledger path after successful F4 — previously + # the multi-session chain bypassed STAGE_REGISTRY's F5 entry entirely, + # so the hand-off report was never produced for --sessions > 1. + if f4_result.status in ("pass", "warn"): + f5_ctx = ChainContext( + persona_id=persona_id, + max_turns=0, + k=0, + out_dir=out_dir, + scale_scores_path=None, + ) + all_results.append(await run_f5_stage(f5_ctx)) + else: + all_results.append( + StageResult( + "F5", + "skip", + "F4 produced no longitudinal output — F5 skipped (dependency not met)", + ) + ) return all_results @@ -1441,12 +1461,14 @@ def _run_f5_report( async def run_f5_stage(ctx: ChainContext) -> StageResult: - """STAGE_REGISTRY entry point (single-session `run_chain` path, right - after F4 — same "reads whatever ledger entries have accumulated" - post-loop role `run_f4_stage` already plays for F4). Delegates entirely - to `_run_f5_report`; maps `F5InsufficientSessionsError` to a "skip" - `StageResult` (same discipline as `_run_f4_analysis`'s own <2-entries - skip), any other exception to a named "fail".""" + """Post-ledger F5 invocation shared by BOTH chain paths (`_main`'s + single-session path after its ledger append, and + `run_multi_session_chain` after a successful F4) — F5 reads the session + ledger, so it must only ever run once the current invocation's entries + are all appended. Delegates entirely to `_run_f5_report`; maps + `F5InsufficientSessionsError` to a "skip" `StageResult` (same + discipline as `_run_f4_analysis`'s own <2-entries skip), any other + exception to a named "fail".""" t0 = time.perf_counter() try: paths = _run_f5_report(ctx.persona_id, ctx.out_dir) @@ -1794,8 +1816,10 @@ async def _main(args: argparse.Namespace) -> int: return 1 ctx.conversation_path = path - results = await run_chain(ctx) - print_report(ctx, results) + # F5 is excluded from the in-chain traversal and invoked from the + # post-ledger path below — inside run_chain it read a ledger that did + # not yet contain THIS session's entry (stale/skipped hand-off report). + results = await run_chain(ctx, stages=[s for s in STAGE_REGISTRY if s.name != "F5"]) # Plan §6 item 5: single-session ledger gap fix. Only written when F1 # itself did not hard-fail (mirrors run_multi_session_chain's own @@ -1806,6 +1830,21 @@ async def _main(args: argparse.Namespace) -> int: ledger_path = _ledger_path(args.persona, out_dir) _append_ledger_entry(ledger_path, _build_single_session_ledger_entry(ctx, results)) + f4_result = next((r for r in results if r.name == "F4"), None) + if any(r.status == "fail" for r in results): + f5_result = StageResult("F5", "skip", "prior stage failed — F5 skipped (dependency)") + elif f4_result is None or f4_result.status not in ("pass", "warn"): + f5_result = StageResult( + "F5", + "skip", + "F4 produced no longitudinal output this run — F5 skipped (dependency not met)", + ) + else: + f5_result = await run_f5_stage(ctx) + f6_index = next((i for i, r in enumerate(results) if r.name == "F6"), len(results)) + results.insert(f6_index, f5_result) + + print_report(ctx, results) return 0 if all(r.status != "fail" for r in results) else 1 diff --git a/apps/ai-server/src/f5.py b/apps/ai-server/src/f5.py index 1cc3a35..3c567af 100644 --- a/apps/ai-server/src/f5.py +++ b/apps/ai-server/src/f5.py @@ -45,6 +45,7 @@ from __future__ import annotations import re +import unicodedata from dataclasses import dataclass, field from datetime import date as _date from datetime import datetime @@ -788,15 +789,22 @@ def _build_a7(inp: HandoffReportInput) -> RecommendationSection: # ── A8 (optional narrative hook, Task 2 / `handoff_generator` v3) ──────── +def _leak_normalize(s: str) -> str: + """NFKC + casefold so the A6→A8 refusal cannot be dodged by case or + unicode-width variants ("ptsd", "ＰＴＳＤ" must match candidate "PTSD").""" + return unicodedata.normalize("NFKC", s).casefold() + + def _build_a8(inp: HandoffReportInput) -> NarrativeSection: """`ADR-037` Decision 1 default (`narrative_enabled=False`) is UNCHANGED — A8 still ships the explicit disabled marker, never blank, whenever the caller does not opt in. Task 2 adds the OPT-IN path: when `narrative_enabled=True` (enforced non-empty `narrative_text`, `assemble_handoff_report` below), this function applies ONE - defense-in-depth check before rendering it — a plain substring scan of - every A6 candidate's `disease` name against the given text (HPI hard - red line, design doc §6.1 point 1). A match REFUSES the narrative + defense-in-depth check before rendering it — an NFKC+casefold + normalized substring scan of every A6 candidate's `disease` name + against the given text (HPI hard red line, design doc §6.1 point 1). + A match REFUSES the narrative entirely (never silently strips/redacts the matched substring, which could leave a mangled sentence that still implies the missing content) — this function still never calls any LLM/agent itself @@ -808,7 +816,8 @@ def _build_a8(inp: HandoffReportInput) -> NarrativeSection: text = (inp.narrative_text or "").strip() apd = inp.domain_inference.ai_predicted_disease candidate_diseases = [c.disease for c in (apd.candidates if apd else []) if c.disease] - leaked = [d for d in candidate_diseases if d in text] + normalized_text = _leak_normalize(text) + leaked = [d for d in candidate_diseases if _leak_normalize(d) in normalized_text] if leaked: return NarrativeSection( narrative_enabled=False, text=None, absent_marker=NARRATIVE_REJECTED_DISEASE_LEAK_KO @@ -865,7 +874,7 @@ def assemble_handoff_report(inp: HandoffReportInput) -> HandoffReportOutput: return HandoffReportOutput( vp_id=inp.vp_id, - generated_at=datetime.now().isoformat(), + generated_at=datetime.now().astimezone().isoformat(), a0_header=_build_header(inp), a1_chief_complaint=_build_a1(inp), a2_hpi=_build_a2(inp), diff --git a/apps/ai-server/src/schemas/handoff.py b/apps/ai-server/src/schemas/handoff.py index 908ae95..5cd2ab2 100644 --- a/apps/ai-server/src/schemas/handoff.py +++ b/apps/ai-server/src/schemas/handoff.py @@ -2,11 +2,65 @@ from __future__ import annotations -from pydantic import BaseModel, Field +from pydantic import BaseModel, ConfigDict, Field, field_validator from src.agents.base import AgentInput, AgentOutput from src.schemas.common import EvidencePacket, RiskLevel +_VALID_CTRS_LABELS = frozenset({"1", "2", "3", "4", "5"}) +_VALID_RISK_LABELS = frozenset(level.value for level in RiskLevel) + + +class RiskEvent(BaseModel): + """A single safety event attached to a handoff request. + + Severity labels are strictly validated at the API boundary (ISS-021 + hardening): an invalid ``risk_level``/``ctrs_level`` is rejected with a + validation error instead of being silently floored to ``medium`` + downstream — e.g. the unicode-confusable CTRS label ``"①"`` (intent: + CTRS 1 = critical) previously passed ``str.isdigit()`` but failed + ``int()``, downgrading a critical event. Extra keys (e.g. ``crisis``, + free-text reasons) are preserved verbatim. An event with NO severity + keys remains valid and floors to medium downstream (issue #21). + """ + + model_config = ConfigDict(extra="allow") + + risk_level: str | None = Field( + default=None, + description="One of: none | low | medium | high | critical", + ) + ctrs_level: str | None = Field( + default=None, + description="Crisis Triage Rating Scale, ASCII digit '1'-'5'", + ) + + @field_validator("risk_level", mode="before") + @classmethod + def _normalize_risk_level(cls, value: object) -> str | None: + if value is None: + # Explicit null is a malformed severity claim — reject. A genuinely + # ABSENT field never reaches this validator (pydantic skips + # validators for unset fields) and keeps the None default. + raise ValueError("risk_level must not be null — omit the field instead") + raw = str(value).strip().lower() + if raw not in _VALID_RISK_LABELS: + raise ValueError( + f"risk_level must be one of {sorted(_VALID_RISK_LABELS)}, got {value!r}" + ) + return raw + + @field_validator("ctrs_level", mode="before") + @classmethod + def _normalize_ctrs_level(cls, value: object) -> str | None: + if value is None: + raise ValueError("ctrs_level must not be null — omit the field instead") + raw = str(value).strip() + # ASCII-strict: rejects unicode digits ("١", "①"), out-of-range, non-digits. + if raw not in _VALID_CTRS_LABELS: + raise ValueError(f"ctrs_level must be an ASCII digit '1'-'5', got {value!r}") + return raw + class SlotData(BaseModel): """Collected clinical slot data from the dialogue session. @@ -47,9 +101,10 @@ class HandoffInput(AgentInput): slots: SlotData = Field(default_factory=SlotData) conversation_history: list[dict[str, str]] = Field(default_factory=list) scale_scores: list[ScaleScore] = Field(default_factory=list) - risk_events: list[dict[str, str]] = Field( + risk_events: list[RiskEvent] = Field( default_factory=list, - description="Safety events during the session", + max_length=100, + description="Safety events during the session (severity labels strictly validated)", ) ocr_documents: list[dict[str, str]] = Field( default_factory=list, diff --git a/apps/ai-server/src/services/f5_report.py b/apps/ai-server/src/services/f5_report.py index 9b64d30..065534b 100644 --- a/apps/ai-server/src/services/f5_report.py +++ b/apps/ai-server/src/services/f5_report.py @@ -26,6 +26,8 @@ import uuid from datetime import datetime from pathlib import Path +from xml.etree import ElementTree +from xml.sax.saxutils import escape as xml_escape from src.f1 import OUTPUT_DIR from src.schemas.handoff_report import ( @@ -123,9 +125,14 @@ def save_f5_result( md_path.write_text(build_markdown_report(report), encoding="utf-8") paths["markdown"] = md_path + # One exporter's failure must never suppress the remaining artifacts — + # a PDF rendering error still leaves clinicians the md + FHIR outputs. pdf_path = out / f"{prefix}_handoff.pdf" - pdf_path.write_bytes(build_pdf_report(report, chart_paths or {})) - paths["pdf"] = pdf_path + try: + pdf_path.write_bytes(build_pdf_report(report, chart_paths or {})) + paths["pdf"] = pdf_path + except Exception: + logger.exception("F5 PDF export failed — markdown/FHIR artifacts continue") fhir_bundle = build_fhir_bundle(report) fhir_path = out / f"{prefix}_handoff_fhir.json" @@ -296,6 +303,30 @@ def _dimension_ko(dimension: str) -> str: return _DIMENSION_KO.get(dimension, dimension) +def _md_inline(text: object) -> str: + """Neutralize markdown/HTML structure in clinical free text rendered + inline: collapsing whitespace runs removes the line starts that forged + headings/blockquotes/table rows require, and `<` escaping defuses raw + HTML — clinician-facing content must render as DATA, never as markup.""" + return " ".join(str(text).replace("<", "<").split()) + + +def _md_cell(text: object) -> str: + return _md_inline(text).replace("|", "\\|") + + +_MD_LINE_STRUCTURE_RE = re.compile(r"^(\s*)([#>|\-+*])", flags=re.MULTILINE) + +_PDF_PARAGRAPH_MAX_LINES = 40 + + +def _md_block(text: str) -> str: + """Multi-line variant for paragraph-preserving fields (A8 narrative): + keeps line breaks but backslash-escapes line-leading markdown tokens + and defuses raw HTML.""" + return _MD_LINE_STRUCTURE_RE.sub(r"\1\\\2", str(text).replace("<", "<")) + + def _truncate( text: str | None, limit: int = 80, *, appendix: _AppendixCollector, label: str ) -> str: @@ -304,10 +335,11 @@ def _truncate( "상세 부록" (detail appendix) entry carrying the SAME full *text* (never `"(상세 아래)"`, which pointed nowhere). `appendix`/`label` are mandatory — every call site owns an `_AppendixCollector` for its - render pass.""" + render pass. Output is markdown-sanitized (`_md_cell`) — call sites + place it in table cells and inline prose.""" if not text: return "" - t = str(text) + t = _md_cell(text) if len(t) <= limit: return t n = appendix.anchor(label, t) @@ -640,9 +672,9 @@ def _mse_lines(a4: MentalStatusSection) -> list[str]: assessable = [d for d in a4.domain_checklist if d.assessable] if not a4.present and not assessable: return ["텍스트 문진 특성상 관찰 기반 MSE는 평가 불가; 대화에서 도출된 소견 없음"] - lines = [f"{a4.label}: {a4.raw_text}"] if a4.present else [] + lines = [f"{a4.label}: {_md_inline(a4.raw_text)}"] if a4.present else [] if assessable: - lines += [f"- {d.domain}: {d.note}" for d in assessable] + lines += [f"- {d.domain}: {_md_inline(d.note)}" for d in assessable] elif a4.present: lines.append("개별 영역(mood/insight 등) 평가는 이 슬롯 특성상 불가") return lines @@ -924,9 +956,9 @@ def build_markdown_report(report: HandoffReportOutput) -> str: lines += [ "## 주호소 및 현병력", "", - f"**주호소**: {a1.text if a1.present else '미수집'}", + f"**주호소**: {_md_inline(a1.text) if a1.present else '미수집'}", "", - f"**현병력**: {a2.text if a2.present else '미수집'}", + f"**현병력**: {_md_inline(a2.text) if a2.present else '미수집'}", "", "### 정신상태검사 (MSE)", "", @@ -960,7 +992,7 @@ def build_markdown_report(report: HandoffReportOutput) -> str: lines.append("") course_bullets = _major_course_bullets(so) if course_bullets: - lines += [f"- {b_}" for b_ in course_bullets] + lines += [f"- {_md_inline(b_)}" for b_ in course_bullets] else: lines.append("- 표시할 주요 경과 변화 없음") lines.append("") @@ -1109,7 +1141,7 @@ def build_markdown_report(report: HandoffReportOutput) -> str: # discipline as AI 참고 정보/A6, never nested inside another section) ── lines += ["## 임상 종합 소견", ""] if a8.narrative_enabled and a8.text: - lines += [f"> {NARRATIVE_ENABLED_LABEL_KO}", "", a8.text, ""] + lines += [f"> {NARRATIVE_ENABLED_LABEL_KO}", "", _md_block(a8.text), ""] else: lines += [a8.absent_marker, ""] @@ -1119,7 +1151,7 @@ def build_markdown_report(report: HandoffReportOutput) -> str: for n, label, text in appendix.entries: lines += [f"**{n}. {label}**", "", text, ""] slot_history_sections = [ - (row.label, " → ".join(row.change_history_full)) + (row.label, " → ".join(_md_inline(v) for v in row.change_history_full)) for row in _slot_table_rows(so) if len(row.change_history_full) >= 2 ] @@ -1343,7 +1375,17 @@ def build_pdf_report( def P(text: str, style: str = "body") -> Paragraph: safe = (text or "").replace("&", "&").replace("<", "<").replace(">", ">") - return Paragraph(safe.replace("\n", "
"), styles[style]) + # Newline-dense values previously exploded into unbounded
runs, + # aborting the whole PDF with a reportlab LayoutError inside + # unsplittable cells — collapse blank-line runs and cap the total. + collapsed: list[str] = [] + for ln in safe.split("\n"): + if not ln.strip() and collapsed and not collapsed[-1].strip(): + continue + collapsed.append(ln) + if len(collapsed) > _PDF_PARAGRAPH_MAX_LINES: + collapsed = collapsed[:_PDF_PARAGRAPH_MAX_LINES] + ["… (이하 생략)"] + return Paragraph("
".join(collapsed), styles[style]) def _table(rows: list[list[str]], font_size: float = 7.5) -> Table: t = Table(rows, hAlign="LEFT") @@ -1719,8 +1761,13 @@ def _local_concept(code: str, text: str) -> dict: def _div(text: str) -> dict: """`Narrative` (status=generated) wrapping free text in the required - xhtml div — used for every `Composition.section.text` below.""" - return {"status": "generated", "div": f"
{text}
"} + xhtml div — used for every `Composition.section.text` below. Clinical + text is XML-escaped: values like "수면 <4h & 불안" must stay well-formed + XHTML, and markup-shaped input must never survive as live markup.""" + return { + "status": "generated", + "div": f"
{xml_escape(text)}
", + } def build_fhir_bundle(report: HandoffReportOutput) -> dict: @@ -1828,7 +1875,24 @@ def add(key: str, resource: dict) -> str: "code": _local_concept("ctrs", "Crisis Triage Rating Scale (session_ctrs, local)"), "subject": {"reference": patient_url}, "effectiveDateTime": a3.current_simulated_date, - "valueInteger": a3.session_ctrs, + # FHIR JSON forbids null primitives — absent CTRS becomes + # dataAbsentReason below instead of "valueInteger": null. + **( + {"valueInteger": a3.session_ctrs} + if a3.session_ctrs is not None + else { + "dataAbsentReason": { + "coding": [ + { + "system": ( + "http://terminology.hl7.org/CodeSystem/data-absent-reason" + ), + "code": "unknown", + } + ] + } + } + ), "note": [ { "text": ( @@ -2291,16 +2355,25 @@ def validate_fhir_bundle(bundle: dict) -> list[str]: f"{rtype} (fullUrl={e.get('fullUrl')}) missing required field '{f}'" ) - def _walk(obj: object) -> None: + def _walk(obj: object, path: str = "bundle") -> None: if isinstance(obj, dict): ref = obj.get("reference") if isinstance(ref, str) and ref.startswith("urn:uuid:") and ref not in full_url_set: violations.append(f"unresolved reference: {ref}") - for v in obj.values(): - _walk(v) + div = obj.get("div") + if isinstance(div, str): + try: + ElementTree.fromstring(div) + except ElementTree.ParseError as exc: + violations.append(f"{path}: Narrative.div is not well-formed XHTML ({exc})") + for k, v in obj.items(): + if v is None: + violations.append(f"{path}.{k}: null values are forbidden in FHIR JSON") + else: + _walk(v, f"{path}.{k}") elif isinstance(obj, list): - for v in obj: - _walk(v) + for i, v in enumerate(obj): + _walk(v, f"{path}[{i}]") _walk(bundle) diff --git a/apps/ai-server/src/services/trend_plotter.py b/apps/ai-server/src/services/trend_plotter.py index 2d8cda9..99b1966 100644 --- a/apps/ai-server/src/services/trend_plotter.py +++ b/apps/ai-server/src/services/trend_plotter.py @@ -21,6 +21,19 @@ import logging from dataclasses import dataclass from datetime import datetime +from pathlib import Path + +_FONT_ASSET_DIR = Path(__file__).resolve().parents[2] / "assets" / "fonts" + + +def _korean_font_candidates() -> list[str]: + """Hangul-capable font files, bundled SHA-pinned assets first so charts + render Korean labels on any host, then common system locations.""" + return [ + str(_FONT_ASSET_DIR / "NotoSansKR-Subset.ttf"), + "/usr/share/fonts/opentype/noto/NotoSansCJK-Regular.ttc", + "/usr/share/fonts/truetype/nanum/NanumGothic.ttf", + ] logger = logging.getLogger(__name__) @@ -199,11 +212,10 @@ def _render_plot( import matplotlib.font_manager as fm import matplotlib.pyplot as plt - # Korean font - for fpath in [ - "/usr/share/fonts/opentype/noto/NotoSansCJK-Regular.ttc", - "/usr/share/fonts/truetype/nanum/NanumGothic.ttf", - ]: + # Korean font — bundled SHA-pinned assets first (portable: macOS/slim + # containers have neither /usr/share path, which silently degraded every + # Hangul label to missing glyphs), system fonts as fallback. + for fpath in _korean_font_candidates(): try: fm.fontManager.addfont(fpath) plt.rcParams["font.family"] = fm.FontProperties(fname=fpath).get_name() diff --git a/apps/ai-server/tests/test_continuous_test_f4.py b/apps/ai-server/tests/test_continuous_test_f4.py index 2a9ef0e..22f831b 100644 --- a/apps/ai-server/tests/test_continuous_test_f4.py +++ b/apps/ai-server/tests/test_continuous_test_f4.py @@ -426,12 +426,15 @@ async def _fake_f3_stage(f2_ctx): "VP-001", n_sessions=2, max_turns=1, k=1, out_dir=tmp_path, scale_scores_path=None, ) - assert results[-1].name == "F4" + # F5 (hand-off) now follows F4 in the post-loop path — F4 sits second + # to last, F5 last. + assert results[-2].name == "F4" + assert results[-1].name == "F5" # F3 was stubbed to "skip" every session -> no "f3" ledger content -> # F4 assembly still runs (>=2 ledger entries exist) but with no # scale data -> "skip"/"warn"/"pass" are all acceptable non-crash # outcomes here; the key assertion is that F4 was invoked at all. - assert results[-1].status in ("pass", "warn", "skip") + assert results[-2].status in ("pass", "warn", "skip") @pytest.mark.asyncio async def test_run_f4_false_skips_post_loop_step( diff --git a/apps/ai-server/tests/test_continuous_test_f5.py b/apps/ai-server/tests/test_continuous_test_f5.py index b717e79..27656e1 100644 --- a/apps/ai-server/tests/test_continuous_test_f5.py +++ b/apps/ai-server/tests/test_continuous_test_f5.py @@ -684,3 +684,180 @@ def test_subprocess_invocation_produces_outputs_in_out_dir(self, tmp_path: Path) assert "F5 hand-off report complete" in result.stdout produced = list((out_dir / persona_id).glob("*_handoff.pdf")) assert len(produced) == 1 + + +class TestF5ChainWiring: + """Round-1 review blocker: with --sessions>1 F5 never ran (registry + bypassed), and with --sessions 1 F5 ran BEFORE the session's own ledger + entry existed. F5 must run from the post-ledger path after F4.""" + + @staticmethod + async def _fake_f1(persona_id, max_turns, followup_from=None, **kwargs): + from src.f1 import F1Result + + session_index = kwargs["session_index"] + return F1Result( + session_id=f"f1_{persona_id}_s{session_index}", + persona_id=persona_id, + persona_name="테스트", + session_index=session_index, + is_revisit=session_index > 1, + model="stub-model", + prompt_version="v3", + final_slots=[{"key": "chief_complaint", "value": "cc"}], + ) + + @pytest.mark.asyncio + async def test_multi_session_runs_f5_after_f4_over_complete_ledger( + self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch + ) -> None: + conv_paths: list[Path] = [] + ledger_len_at_f5_call: list[int] = [] + + async def _fake_run_simulation(persona_id, max_turns, followup_from=None, **kwargs): + session_index = kwargs["session_index"] + conv_paths.append(_write_conversation(tmp_path, persona_id, session_index)) + return await TestF5ChainWiring._fake_f1( + persona_id, max_turns, followup_from, **kwargs + ) + + async def _fake_run_f2_stage(f2_ctx): + return ct.StageResult("F2", "pass", "ok") + + async def _fake_f4(persona_id, out_dir): + return ct.StageResult("F4", "pass", "ok") + + async def _fake_f5(f5_ctx): + entries = json.loads( + ct._ledger_path(f5_ctx.persona_id, f5_ctx.out_dir).read_text(encoding="utf-8") + ) + ledger_len_at_f5_call.append(len(entries)) + return ct.StageResult("F5", "pass", "ok") + + import src.f1 as f1_module + + monkeypatch.setattr(f1_module, "_run_simulation", _fake_run_simulation) + monkeypatch.setattr(ct, "run_f2_stage", _fake_run_f2_stage) + monkeypatch.setattr(ct, "_find_latest_f1_conversation", lambda persona_id: conv_paths[-1]) + monkeypatch.setattr(ct, "_run_f4_analysis", _fake_f4) + monkeypatch.setattr(ct, "run_f5_stage", _fake_f5) + + results = await ct.run_multi_session_chain( + "VP-W1", + n_sessions=2, + max_turns=3, + k=3, + out_dir=tmp_path, + scale_scores_path=None, + answer_mode="expected", + run_f4=True, + ) + names = [r.name for r in results] + assert "F5" in names, f"F5 stage never ran in multi-session chain: {names}" + assert names.index("F5") > names.index("F4") + assert ledger_len_at_f5_call == [2], ( + f"F5 must run over the COMPLETE 2-entry ledger, saw {ledger_len_at_f5_call}" + ) + + @pytest.mark.asyncio + async def test_multi_session_skips_f5_when_f4_fails( + self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch + ) -> None: + conv_paths: list[Path] = [] + + async def _fake_run_simulation(persona_id, max_turns, followup_from=None, **kwargs): + session_index = kwargs["session_index"] + conv_paths.append(_write_conversation(tmp_path, persona_id, session_index)) + return await TestF5ChainWiring._fake_f1( + persona_id, max_turns, followup_from, **kwargs + ) + + async def _fake_run_f2_stage(f2_ctx): + return ct.StageResult("F2", "pass", "ok") + + async def _fake_f4_fail(persona_id, out_dir): + return ct.StageResult("F4", "fail", "boom") + + f5_calls: list[str] = [] + + async def _fake_f5(f5_ctx): + f5_calls.append(f5_ctx.persona_id) + return ct.StageResult("F5", "pass", "ok") + + import src.f1 as f1_module + + monkeypatch.setattr(f1_module, "_run_simulation", _fake_run_simulation) + monkeypatch.setattr(ct, "run_f2_stage", _fake_run_f2_stage) + monkeypatch.setattr(ct, "_find_latest_f1_conversation", lambda persona_id: conv_paths[-1]) + monkeypatch.setattr(ct, "_run_f4_analysis", _fake_f4_fail) + monkeypatch.setattr(ct, "run_f5_stage", _fake_f5) + + results = await ct.run_multi_session_chain( + "VP-W1B", + n_sessions=1, + max_turns=3, + k=3, + out_dir=tmp_path, + scale_scores_path=None, + answer_mode="expected", + run_f4=True, + ) + f5_results = [r for r in results if r.name == "F5"] + assert f5_calls == [], "F5 must not execute when F4 failed" + assert f5_results and f5_results[0].status == "skip" + + @pytest.mark.asyncio + async def test_single_session_runs_f5_after_ledger_append( + self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch + ) -> None: + import argparse + + conv_path = _write_conversation(tmp_path, "VP-W1S", 1) + ledger_len_at_f5_call: list[int] = [] + captured: dict = {} + + async def _fake_run_chain(ctx, stages=None): + ctx.conversation_path = conv_path + stage_names = [s.name for s in (stages if stages is not None else ct.STAGE_REGISTRY)] + return [ + ct.StageResult(n, "pass" if n != "F6" else "skip", "ok") for n in stage_names + ] + + async def _fake_f5(f5_ctx): + entries = json.loads( + ct._ledger_path(f5_ctx.persona_id, f5_ctx.out_dir).read_text(encoding="utf-8") + ) + ledger_len_at_f5_call.append(len(entries)) + return ct.StageResult("F5", "pass", "ok") + + monkeypatch.setattr(ct, "run_chain", _fake_run_chain) + monkeypatch.setattr(ct, "run_f5_stage", _fake_f5) + monkeypatch.setattr( + ct, "print_report", lambda ctx, results: captured.update(results=results) + ) + + args = argparse.Namespace( + persona="VP-W1S", + sessions=1, + max_turns=3, + k=3, + out=str(tmp_path), + scale_scores=None, + start_from_conversation=None, + answer_mode="expected", + force_questionnaire=None, + patient_sex=None, + scenario_pack=None, + no_f4=False, + f5_from_artifacts=None, + session_interval_days=14, + ) + rc = await ct._main(args) + assert rc == 0 + assert ledger_len_at_f5_call == [1], ( + "F5 must run AFTER the single-session ledger entry is appended, " + f"saw ledger lengths {ledger_len_at_f5_call}" + ) + names = [r.name for r in captured["results"]] + assert "F5" in names and "F6" in names + assert names.index("F5") < names.index("F6") diff --git a/apps/ai-server/tests/test_f5.py b/apps/ai-server/tests/test_f5.py index bab09a6..f2c5161 100644 --- a/apps/ai-server/tests/test_f5.py +++ b/apps/ai-server/tests/test_f5.py @@ -258,6 +258,50 @@ def test_disease_leak_is_rejected_not_rendered(self) -> None: assert out.a8_narrative.text is None assert out.a8_narrative.absent_marker == NARRATIVE_REJECTED_DISEASE_LEAK_KO + def test_leak_guard_is_case_insensitive(self) -> None: + # Adversarial: A6 candidate "PTSD" must also be caught as "ptsd". + from src.schemas.handoff_report import NARRATIVE_REJECTED_DISEASE_LEAK_KO + + apd = AIPredictedDiseaseOutput( + candidates=[ + AIPredictedDiseaseCandidate( + disease="PTSD", similarity_score=0.5, source_id="case_card:1", quote="q" + ) + ], + mode="rag_live", + ) + out = assemble_handoff_report( + _build_input( + ai_predicted_disease=apd, + narrative_enabled=True, + narrative_text="환자에게서 ptsd 소견이 의심됨.", + ) + ) + assert out.a8_narrative.narrative_enabled is False + assert out.a8_narrative.absent_marker == NARRATIVE_REJECTED_DISEASE_LEAK_KO + + def test_leak_guard_normalizes_fullwidth_unicode(self) -> None: + # Adversarial: fullwidth "ＰＴＳＤ" NFKC-normalizes to "PTSD" and must be caught. + from src.schemas.handoff_report import NARRATIVE_REJECTED_DISEASE_LEAK_KO + + apd = AIPredictedDiseaseOutput( + candidates=[ + AIPredictedDiseaseCandidate( + disease="PTSD", similarity_score=0.5, source_id="case_card:1", quote="q" + ) + ], + mode="rag_live", + ) + out = assemble_handoff_report( + _build_input( + ai_predicted_disease=apd, + narrative_enabled=True, + narrative_text="환자에게서 ＰＴＳＤ 소견이 의심됨.", + ) + ) + assert out.a8_narrative.narrative_enabled is False + assert out.a8_narrative.absent_marker == NARRATIVE_REJECTED_DISEASE_LEAK_KO + def test_text_without_any_candidate_disease_name_is_not_rejected(self) -> None: clean_text = "환자는 수면 문제와 무기력감을 자가보고함. 위험 관련 소견은 A3 참조." out = assemble_handoff_report( diff --git a/apps/ai-server/tests/test_f5_fhir.py b/apps/ai-server/tests/test_f5_fhir.py index 85f5ea4..76dc450 100644 --- a/apps/ai-server/tests/test_f5_fhir.py +++ b/apps/ai-server/tests/test_f5_fhir.py @@ -615,3 +615,66 @@ def test_bundle_still_validates_structurally_with_a8_enabled(self) -> None: report = _report(narrative_enabled=True, narrative_text="환자는 수면 문제를 자가보고함.") bundle = build_fhir_bundle(report) assert validate_fhir_bundle(bundle) == [] + + +class TestFhirPrimitiveAndNarrativeValidity: + """Round-1 review blockers: naive timestamps, null primitives, unescaped XHTML.""" + + def test_bundle_timestamp_and_composition_date_are_timezone_aware(self) -> None: + from datetime import datetime + + bundle = build_fhir_bundle(_report()) + assert datetime.fromisoformat(bundle["timestamp"]).tzinfo is not None + comp = bundle["entry"][0]["resource"] + assert datetime.fromisoformat(comp["date"]).tzinfo is not None + + def test_absent_ctrs_omits_value_integer_with_data_absent_reason(self) -> None: + bundle = build_fhir_bundle(_empty_report()) + ctrs_obs = [ + e["resource"] + for e in bundle["entry"] + if e["resource"].get("resourceType") == "Observation" + and any(c.get("code") == "ctrs" for c in e["resource"]["code"].get("coding", [])) + ] + assert ctrs_obs, "CTRS observation missing entirely" + obs = ctrs_obs[0] + assert "valueInteger" not in obs + assert obs.get("dataAbsentReason"), "absent CTRS must carry dataAbsentReason" + + def test_narrative_divs_are_wellformed_xml_with_escaped_clinical_text(self) -> None: + import xml.etree.ElementTree as ET + + session = SessionSnapshot( + session_id="f1_VP-XML", + persona_id="VP-XML", + persona_name="김검증", + session_index=1, + simulated_date="2026-01-01", + model="solar-pro3", + final_slots={ + "chief_complaint": "불안 & 수면 <3시간, 기록: ", + }, + session_ctrs=3, + crisis_triggered=False, + crisis_turn=None, + risk_floor=None, + probe_event_count=0, + ) + inp = HandoffReportInput( + vp_id="VP-XML", + session=session, + current_session_f3=None, + all_f3_administrations=(), + domain_inference=DomainInferenceSnapshot( + ai_predicted_disease=AIPredictedDiseaseOutput( + candidates=[], mode="experimental_unpopulated" + ) + ), + longitudinal=LongitudinalAnalysisOutput(vp_id="VP-XML", n_sessions=1), + ) + bundle = build_fhir_bundle(assemble_handoff_report(inp)) + comp = bundle["entry"][0]["resource"] + for section in comp["section"]: + div = section["text"]["div"] + ET.fromstring(div) + assert "", + "history_of_present_illness": "수면 문제\n\n## 위조된 진료 지시\n복약 중단", + "family_history": "약물A | 약물B | 약물C | 약물D | 약물E", + }, + session_ctrs=3, + crisis_triggered=False, + crisis_turn=None, + risk_floor=None, + probe_event_count=0, + ) + inp = HandoffReportInput( + vp_id="VP-INJ", + session=session, + current_session_f3=None, + all_f3_administrations=(), + domain_inference=DomainInferenceSnapshot( + ai_predicted_disease=AIPredictedDiseaseOutput( + candidates=[], mode="experimental_unpopulated" + ) + ), + longitudinal=LongitudinalAnalysisOutput(vp_id="VP-INJ", n_sessions=1), + **kw, + ) + return assemble_handoff_report(inp) + + def test_clinical_text_cannot_forge_headings(self) -> None: + md = build_markdown_report(self._adversarial_report()) + assert not any(line.startswith("## 위조된") for line in md.splitlines()) + + def test_raw_html_is_neutralized(self) -> None: + md = build_markdown_report(self._adversarial_report()) + assert "