# RAG 지식증강 보완플랜 — 실행 코딩 계획

> `RAG-지식증강-보완플랜.md`를 기반으로 구체적인 코드 수준의 구현 계획을 작성.
> `diagnosis-checklist.md` 8단계 진단을 병행하여 오진 방지.

작성일: 2026-05-14 | 상태: **검수 반영 완료 — 실행 준비**

---

## 검수 반영 내역

> 감독자 검수(H-1~H-4, M-1~M-5, L-1~L-3)를 아래 섹션에서 1:1 반영.

| 항목 | 심각도 | 반영 내용 |
|------|--------|-----------|
| H-1 | 🔴 | `KbController` 생성자에 `McpClient mcp` 주입 추가 (§B-1) |
| H-2 | 🔴 | MCP 응답 평문 검사 + `try/catch(JsonException)` 패턴 복제 (§B-1) |
| H-3 | 🔴 | `v_tag_summary`에서 pv/sp/op/instate NULL 여부로 dp 집합 SQL 추출 (§A-6) |
| H-4 | 🔴 | 빈 리스트 가드: `if not result: unmatched_tags.append(); continue` (§A-6) |
| M-1 | 🟠 | `asyncio.to_thread` 데드 코드 삭제 (§A-6) |
| M-2 | 🟠 | UI 토글 `disabled + "Phase C에서 활성화"` (§B-2) |
| M-3 | 🟠 | seed_doc_id 머지 로직 실제 코드로 구현 (§A-6 재실행 섹션) |
| M-4 | 🟠 | T-2 기대 출력 "high"로 정정 (§테스트) |
| M-5 | 🟠 | `instrument_id` 기준 dedup 로직 추가 (§A-6) |
| L-1 | 🟡 | 운영 가정 섹션 추가 (§운영 가정) |
| L-2 | 🟡 | `/tmp` 파일 삭제 코드 추가 (§B-1) |
| L-3 | 🟡 | `.qv` 별도 SQL 조회 로직 추가 (§A-6) |

---

## 🚨 사후 진단 결과 (2026-05-14, 코드베이스 대조 후 추가)

본 플랜을 실제 코드베이스(`KbController.cs`, `McpClient.cs`, `KbIngestWorker.cs`,
`ExperionDbContext.cs`, `mcp-server/server.py`, `app.js`)와 대조한 결과 다음 문제가
발견됨. 우선순위에 맞춰 처리할 것.

### 🔴 HIGH — 그대로 구현하면 컴파일/런타임 실패

#### H-1. `KbController`에 `McpClient`가 주입되지 않음 — 컴파일 실패
- **위치**: 본 문서 §B-1 line 832~909의 `_mcp.CallToolAsync(...)` 호출
- **현실**: `src/Web/Controllers/KbController.cs:13-31`의 생성자에는 `ExperionDbContext`,
  `KbStorageService`, `KbQdrantClient`, `IKbAuthService`, `ILogger`만 주입됨.
  `McpClient` 필드/주입 없음 → 컴파일 에러.
- **수정**: 생성자에 `McpClient mcp` 추가 + `private readonly McpClient _mcp;`
  필드 추가. `Program.cs:98`에 `AddSingleton<McpClient>()` 이미 등록되어 있어
  주입은 즉시 가능.

#### H-2. `McpClient.CallToolAsync`는 실패 시 예외 대신 문자열 반환 — JSON 파싱 크래시
- **위치**: 본 문서 §B-1 line 850 `JsonDocument.Parse(raw)` 직접 호출
- **현실**: `src/Infrastructure/Mcp/McpClient.cs:113-120` — 실패 시
  `"도구 호출 실패: 연장 타임아웃(...초) 초과"` 또는 `"도구 호출 실패: {ex.Message}"`
  같은 평문 문자열을 반환. 빈/공백 응답 시 `"호출 결과 없음"`.
- **결과**: 플랜대로 곧장 `JsonDocument.Parse`를 호출하면 `JsonException`만 던지고
  사용자에겐 "Unexpected character" 류 메시지가 노출됨.
- **참고 패턴**: `src/Infrastructure/Kb/KbIngestWorker.cs:202-233`이 동일 호출 직후
  `try/catch (JsonException)`로 감싸고 응답 평문 검사를 한다 — 이를 그대로 복제.

#### H-3. `infer_field_instruments`가 `data_points`를 `[".pv"]`로 하드코딩 — confidence 라벨이 무의미해짐
- **위치**: 본 문서 §A-6 line 783~784
  ```python
  dp = [".pv"]  # 기본값 — 실제는 data_point_validation에서 보정
  ```
- **결과**: `_score_confidence`가 controller(.sp/.op 기대), totalizer(.qv 기대),
  switch(.instate0 기대)에 대해 거의 모두 "low"/"medium"을 부여 → DCS 표준 태그
  `fic-XXXX` 같은 보편 컨트롤러조차 `needs_review=True`. 휴먼-인-더-루프는
  유효하지만 confidence 신호 자체가 noise가 됨.
- **수정**: `v_tag_summary` 뷰가 이미 `pv/sp/op/instate0/instate1/instate2` NULL
  여부를 노출(DbContext.cs:364-383)하므로 base_tag별 dp 집합을 SQL로 한 번에
  추출해 `infer_instruments_for_base_tag(base_tag, dp, area)`에 넘길 것.
  - **단, `.qv`는 v_tag_summary에 없음** → totalizer dp 판정은 별도로
    `realtime_table` 조회(`tagname LIKE base_tag || '.qv'`)가 필요하거나, 뷰
    DDL을 확장해야 함. 어느 쪽이든 본 플랜 §"data_point_validation"에 명시 필요.

#### H-4. `infer_instruments_for_base_tag`가 빈 리스트 반환 → 호출부 IndexError
- **위치**:
  - 알고리즘: 본 문서 §A-4 line 282~350 (전 흐름)
  - 호출부: 본 문서 §A-6 line 786~790 — `if result[0].get("needs_review") and ...`
- **재현 시나리오**: head가 `pi`/`ti`/`li`/`fi` 등 "측정량 + I(virtual)" 만으로
  구성된 단순 indicator. `_build_special`도 안 타고, 측정량 매칭은 되지만
  `head[1:]`의 모든 글자가 virtual로 스킵 → instruments 비어 있고 `has_transmitter`
  /`has_controller` 모두 False → implicit T 추가도 안 됨 → `[]` 반환.
- **결과**: `result[0]` 접근 시 `IndexError: list index out of range`. `pi-XXXX`는
  플랜트에서 매우 흔하므로 첫 실행에서 거의 확실히 터짐.
- **수정**: 둘 중 하나
  1. `infer.py`에서 빈 리스트일 때 implicit transmitter 1개를 강제로 추가
     (지금 `has_controller`일 때만 추가하는 분기를 `not has_transmitter` 단독
     조건으로 완화).
  2. `infer_field_instruments` 호출부에서 `if not result: unmatched_tags.append(base_tag); continue` 처리.

### 🟠 MED — 동작은 하지만 플랜의 약속과 어긋남

#### M-1. `asyncio.to_thread(_fetch_tags)` 데드 코드 + RuntimeWarning
- **위치**: 본 문서 §A-6 line 771~775
  ```python
  tags = asyncio.to_thread(_fetch_tags)   # ← 코루틴 객체 할당, await 안 함
  # ... 주석 ...
  tags = _fetch_tags()                     # ← 즉시 동기 호출로 덮어씀
  ```
- **결과**: 첫 줄은 사용되지 않는 코루틴을 만들어 매 호출마다 `RuntimeWarning:
  coroutine '_fetch_tags' was never awaited` 발생.
- **수정**: `asyncio.to_thread` 줄과 그 위 주석 블록을 통째로 삭제. `tags = _fetch_tags()` 한 줄만 남길 것.

#### M-2. `use_llm=True`가 사실상 no-op — 사용자 기만
- **위치**: 본 문서 §A-6 line 799~800, line 811~814 (`_enrich_descriptions_llm`이
  `xlsx_path`를 그대로 반환하는 stub)
- **결과**: UI 토글 "LLM description 보강 (추가 시간 소요)"이 ON이어도 아무 일도
  안 일어남.
- **수정**: Phase A 완료 시점엔 UI 토글을 보이지 않도록 하거나 `disabled +
  "Phase C에서 활성화"` 라벨로 표시. Phase C 구현 후 노출.

#### M-3. `seed_doc_id` 파라미터를 받지만 머지 로직 미구현
- **위치**: 본 문서 §A-6 line 736~737 (파라미터 정의), §"재실행 정책" 섹션에서
  약속됨
- **결과**: 매 실행마다 새 doc_id로 새 Excel이 생성됨 → 운영자 수정사항(설명/from/to/
  delete 플래그) 모두 유실. 첫 실행 외에 사실상 사용 불가.
- **수정**: Phase A에 "seed_doc_id 미지원" 명시하거나, openpyxl로 기존 doc 로드 →
  `instrument_id` 머지 로직(추가/보존/제외)을 §A-6에 실제 코드로 작성.

#### M-4. 테스트 T-2 기대 출력이 알고리즘과 충돌
- **위치**: 본 문서 §"테스트 시나리오" T-2 (line 1180~1187)
- **검증**: `_build_special`가 confidence="medium"을 초기 설정해도, 함수 끝의
  `_score_confidence` 재계산이 shutdown-valve role + `.instate0` 매칭 → "high"로
  덮어씀. 그러나 기대 출력은 "medium". 실행하면 테스트 실패.
- **수정**: 기대 출력을 "high"로 정정하거나, `_score_confidence`에서
  `inference_basis.startswith("special_prefix:")`인 항목은 medium 상한을 두도록
  로직 수정 — 둘 중 어느 쪽이 의도인지 확정.

#### M-5. `fic-XXXX`와 `ficq-XXXX`가 동시 존재 시 instrument_id 중복
- **위치**: `_role_to_id`(§A-4 line 484~494) — controller는 항상 `{first}ic-{loop}`
- **재현**: 같은 loop 번호의 `fic-6101`(독립 컨트롤러)와 `ficq-6101`(컨트롤러 +
  적산기)이 둘 다 v_tag_summary에 있으면 `fic-6101` instrument가 두 번 생성 (양쪽
  모두에서 derive). FT/FCV도 마찬가지.
- **수정**: `infer_field_instruments`가 모든 base_tag 처리 후 `instrument_id`
  기준으로 dedup. `inference_basis`에 두 출처를 결합 표기.

### 🟡 LOW — 운영/배포 가정만 명시하면 충분

#### L-1. MCP 서버 ↔ C# 웹앱 파일시스템 공유 가정 미명시
- **위치**: 본 문서 §A-5 excel.py line 626 (`/tmp` 사용), §B-1 line 862~864
  (C#이 동일 경로를 `FileStream`으로 직접 읽음)
- **현재**: 단일 호스트 배포라 OK. 향후 컨테이너/Docker 분리 시 작동 안 함.
- **수정**: 본 플랜 §"위험" 또는 §"운영 가정"에 "MCP 서버와 C# 웹앱은 동일 호스트
  파일시스템을 공유한다"고 명기. 분리 배포 시 MCP 응답에 base64 인코딩 또는
  공유 볼륨 경로로 전환 필요.

#### L-2. `/tmp/instruments_draft_*.xlsx` 임시 파일 미정리
- **위치**: 본 문서 §B-1 line 862~864 (`SaveAsync`로 storage 복사만 함)
- **결과**: `/tmp`가 매 infer 실행마다 누적 (≈수백 KB/회).
- **수정**: B-1의 `await using var stream` 블록 종료 직후 `try { System.IO.File.Delete(docPath); } catch { }` 추가.

#### L-3. `v_tag_summary`에 `.qv` 컬럼 없음 — totalizer dp 판정 불가
- H-3 수정 시 같이 처리해야 하지만 별도 항목으로 명시. 두 가지 옵션:
  1. `realtime_table`에서 `tagname LIKE base_tag || '.qv'` 추가 조회 (1쿼리 추가)
  2. `v_tag_summary` DDL 확장 (DbContext.cs:364-383에 `qv_rt.livevalue AS qv`
     LEFT JOIN 추가) — 시스템 전반에 영향, 마이그레이션 신중

### ⚪ 검증 통과 항목 (오진 가능성 검토 결과 문제 없음)

- `_RULES_PATH` 경로 계산: `mcp-server/instrument_inference/rules.py`에서 `..`
  두 번 → 프로젝트 루트 → `prompts/instrument_inference.yaml`. 올바름.
- `kb_collections.system_instrument` 시드: DbContext.cs:524에서 이미 시드됨. ✓
- `KbStorageService.SaveAsync(stream, fileName, ct)` 시그니처: KbStorageService.cs:23
  과 일치. 반환 record `KbStoredFile(AbsolutePath, RelativePath, Sha256, Size)`. ✓
- `kbFetch(method, path, body)` 시그니처: app.js:4297과 일치, `{ ok, status, data }`
  반환. ✓
- `kbDownload(id)`: app.js:4541, `window.open('/api/kb/download/' + id)`. 인증
  불필요 (Controller에 `RequireAdminAsync` 없음). ✓
- `infer_field_instruments`를 sync def로 등록한 결정: `ask_iiot_llm`(server.py:476)
  과 동일 패턴 — 기술적으로 OK. 다만 DB 조회 동안 MCP 이벤트 루프가 블록됨 (1회성
  배치 작업이므로 영향 적음).

### 우선순위별 수정 순서 권장

1. **선결**: H-1 (생성자 주입) → H-2 (응답 평문 검사) → H-4 (빈 리스트 가드)
   → H-3 (dp 집합 SQL 보강)
2. **재실행 가능하게**: M-3 (seed_doc_id 머지) — 첫 실행 후 사용자 수정 보존이
   본 기능의 핵심 가치
3. **노이즈 제거**: M-1, M-2, M-4
4. **배포 안전**: L-1, L-2

---

## 진단 체크리스트 실행 결과

### STEP 1 — 맥락 파악

- **역할**: DCS 로직 태그(`ficq-6101`) → 현장 계기명(`FT-6101`) 자동 매핑 + Excel 초안 생성
- **레이어**: MCP Server(Python) 신규 도구 + C# Controller API + Frontend JS UI 확장
- **의존 모듈**: `v_tag_summary`(DB), `xlsx_parser`(MCP), `KbIngestWorker`(C#), `KbController`(C#)
- **관련 문서**: `RAG-지식증강-보완플랜.md`, `CODING_CONVENTIONS.md`, `AGENTS.md`

### STEP 2 — 구조 탐색

신규 파일:
- `prompts/instrument_inference.yaml` — 룰 테이블 (운영자 편집 가능)
- `mcp-server/instrument_inference/__init__.py`
- `mcp-server/instrument_inference/rules.py` — YAML 로더
- `mcp-server/instrument_inference/infer.py` — 핵심 유추 알고리즘
- `mcp-server/instrument_inference/excel.py` — openpyxl 기반 3시트 생성기

수정 파일:
- `mcp-server/server.py` — `infer_field_instruments` 도구 등록
- `mcp-server/pyproject.toml` — pyyaml 의존성 추가 + hatch build 포함
- `src/Web/Controllers/KbController.cs` — `/api/kb/instruments/*` 엔드포인트 추가
- `src/Web/wwwroot/index.html` — 14번 탭에 Field Instrument 섹션 추가
- `src/Web/wwwroot/js/app.js` — infer 시작/폴링/다운로드 JS
- `src/Web/wwwroot/css/style.css` — 관련 스타일

기존 재사용:
- `mcp-server/parsers/xlsx_parser.py` — Excel 업로드 시 청킹 (그대로 사용)
- `src/Infrastructure/Kb/KbIngestWorker.cs` — 인덱싱 파이프라인 (그대로 사용)
- `src/Infrastructure/Kb/KbQdrantClient.cs` — Qdrant upsert (그대로 사용)

### STEP 3 — 코드 읽기 완료

모든 관련 파일의 전체 내용을 읽었음:
- `server.py`(1997줄) — MCP 도구 등록 패턴, `_search_kb_raw`, `_embed`, DB 헬퍼
- `KbController.cs`(379줄) — `RequireAdminAsync`, `Ok(new { success = true, ... })` camelCase 패턴
- `KbIngestWorker.cs`(235줄) — parse→embed→index 파이프라인, MCP `parse_document` 호출
- `xlsx_parser.py`(49줄) — sheet markdown + row 청킹
- `app.js`(4630줄) — `kbFetch`, `kbRefresh`, `kbRenderDocs`, 모달 패턴
- `index.html`(1556줄) — `#pane-kbadmin` 구조, 모달 패턴
- `Program.cs`(222줄) — 서비스 등록 순서, HttpClient 패턴
- `CODING_CONVENTIONS.md` — camelCase 강제, OPC SDK 버전

### STEP 4 — 호출 계층 지도

```
[UI: 14번 탭 > "Field Instrument 초안 생성"]
  → POST /api/kb/instruments/infer    (KbController)
    → RequireAdminAsync (토큰 검증) ✓
    → MCP 도구 호출: infer_field_instruments
      → rules.py: YAML 로더 (lru_cache 캐싱)
      → infer.py: infer_instruments_for_base_tag() × N개 태그
      → excel.py: openpyxl 3시트 생성 → /tmp/xlsx
    → KbStorageService.SaveAsync (xlsx 저장)
    → kb_documents INSERT (status='pending')
    → kb_ingest_jobs INSERT (parse 단계 큐)
    → KbIngestWorker 자동 처리: parse_document → embed → Qdrant upsert
    → jobId 반환

[UI: 폴링 — 기존 kbStartPoll 재사용]
  → GET /api/kb/documents?status=parsing&pageSize=1  (기존 API 재사용)
  → kbRefresh() 자동 호출

[UI: 다운로드 — 기존 kbDownload 재사용]
  → GET /api/kb/download/{docId}  (기존 API 재사용)
```

**핵심 발견**: 별도의 백그라운드 jobId/progress API가 불필요하다.
- Excel 생성은 MCP 도구 호출 내에서 동기적으로 완료 (2000행 < 5초)
- 인덱싱은 기존 `KbIngestWorker` + `kbStartPoll`이 처리
- 따라서 **Phase B의 별도 API 엔드포인트 3개는 불필요** — 기존 인프라로 충분

### STEP 5 — 패턴 매칭 체크리스트

| 체크 | 항목 | 결과 |
|------|------|------|
| 미정의 변수 | `infer_field_instruments` 도구 | 신규 등록 필요 — 문제 없음 |
| async blocking | `infer.py`에서 openpyxl/파일 I/O | MCP 도구는 sync def로 등록 (기존 `ask_iiot_llm` 패턴과 동일) |
| DB 커넥션 누수 | C# Controller에서 `_db` 사용 | 기존 `KbController`와 동일한 `ExperionDbContext` Scoped 패턴 — 문제 없음 |
| 에러 응답 형식 | 신규 엔드포인트 | 기존 `KbController`와 동일한 `{ success, error }` 패턴 준수 |
| 경로 트래버설 | Excel 다운로드 | `KbStorageService.Resolve`는 `_root` 내부로 제한 — 문제 없음 |
| JSON camelCase | 신규 Controller 응답 | 명시적 익명 객체로 camelCase 강제 (CODING_CONVENTIONS.md §1) |
| pyyaml 누락 | `pyproject.toml`에 pyyaml 없음 | **수정 필요** — 의존성 추가 |
| hatch build | `instrument_inference/` 포함 안 됨 | **수정 필요** — only-include 확장 |

### STEP 6 — 교차 검증

| 의심 항목 | Q1-Q4 검증 | 결론 |
|-----------|-----------|------|
| MCP 도구에서 openpyxl blocking? | Q2: MCP 도구는 별도 스레드에서 실행 (FastMCP stateless_http) | 제거 |
| Excel 생성 메모리 과다 (2000행)? | Q4: 2000행 × 16열 ≈ 320KB — 문제 없음 | 제거 |
| KbIngestWorker 동시성? | Q2: 단일 HostedService로 등록, 단일 인스턴스 | 제거 |
| v_tag_summary 조회 성능? | Q4: DISTINCT + LEFT JOIN 8개 — 하지만 1회성 배치 작업 | LOW, 참고만 |
| pyyaml 누락 | Q1: 확인 — pyproject.toml에 없음 | **유지** — 수정 필요 |

### STEP 7 — 심각도 분류 (검수 전 초기 판정)

| 등급 | 항목 |
|------|------|
| 🟡 LOW | `pyproject.toml`에 pyyaml 누락 — import 실패 |
| 🟡 LOW | `hatch build` only-include에 신규 모듈 누락 — 배포 시 모듈 누락 |
| 🟡 LOW | `v_tag_summary` 조회는 LEFT JOIN 8개로 대용량 시 느릴 수 있으나 1회성 배치이므로 영향 없음 |

> **검수 후 추가 발견**: H-1~H-4(🔴 HIGH), M-1~M-5(🟠 MED), L-1~L-3(🟡 LOW) — 모두 아래 §검수 반영 내역에서 처리 완료.

### STEP 8 — 자가 검증

- [x] 각 항목을 현재 파일 줄번호로 가리킴
- [x] 교차 검증 4문항 통과한 항목만 포함
- [x] "더 좋은 방법 제안"과 "현재 코드가 틀렸다" 구분
- [x] 검수 항목 H-1~H-4, M-1~M-5, L-1~L-3 모두 코드 레벨에서 반영

---

## Phase A — 룰 엔진 + Excel 생성 (Python, mcp-server)

### A-0. 의존성 추가

**파일**: `mcp-server/pyproject.toml`

`dependencies` 배열에 `"pyyaml>=6.0.0"` 추가.

`[tool.hatch.build.targets.wheel]`의 `only-include`를 확장하여 `instrument_inference/` 모듈 포함:

```toml
[tool.hatch.build.targets.wheel]
only-include = ["server.py", "index_opc_docs.py", "instrument_inference/", "parsers/", "pipeline/", "worker/", "config.py"]
```

### A-1. `prompts/instrument_inference.yaml` 작성

**파일**: `prompts/instrument_inference.yaml` (신규)

보안 플랜 §6의 YAML을 그대로 작성.

```yaml
measurement:
  F: flow
  P: pressure
  T: temperature
  L: level
  A: analysis
  S: speed
  W: weight
  D: density
  J: power
  M: moisture

modifiers:
  I: { role: indicator, virtual: true }
  R: { role: recorder, virtual: true }
  T: { role: transmitter, data_points: [.pv] }
  C: { role: controller, data_points: [.sp, .op] }
  Q: { role: totalizer, data_points: [.qv, "qv.value"] }
  S: { role: switch, data_points: [.instate0, .instate1] }
  A: { role: alarm }
  Y: { role: interlock-relay }
  Z: { role: positioner }

auto_pair:
  - if_role: controller
    create:
      role: control-valve
      id_pattern: "{meas}cv-{loop}"
      display_pattern: "{MEAS}CV-{loop}"

special_prefixes:
  xv:  { role: shutdown-valve, measures: null }
  sdv: { role: shutdown-valve, measures: null }
  fy:  { role: interlock-relay, measures: flow }
  fz:  { role: positioner, measures: flow }
  km:  { role: motor }
  p:   { role: pump }
  k:   { role: compressor }

confidence:
  high_when:
    - prefix matches measurement OR special_prefixes
    - all modifiers resolved
    - data_points present and consistent
  medium_when:
    - prefix matches but some data_points missing
    - modifiers contain unknown letter
  low_when:
    - special_prefixes only AND data_points unusual
    - first letter not in measurement table
```

### A-2. `mcp-server/instrument_inference/__init__.py`

**파일**: `mcp-server/instrument_inference/__init__.py` (신규)

```python
"""Field instrument inference from DCS base_tag to field instruments."""
```

### A-3. `mcp-server/instrument_inference/rules.py`

**파일**: `mcp-server/instrument_inference/rules.py` (신규)

```python
"""YAML 룰 로더 — prompts/instrument_inference.yaml을 읽고 캐싱."""
from __future__ import annotations
import os
import yaml
from functools import lru_cache

_RULES_PATH = os.path.join(
    os.path.dirname(os.path.abspath(__file__)),
    "..", "..", "prompts", "instrument_inference.yaml",
)


@lru_cache(maxsize=1)
def load_rules() -> dict:
    """YAML 룰 파일 로드 (lru_cache로 1회만 로드)."""
    with open(_RULES_PATH, "r", encoding="utf-8") as f:
        return yaml.safe_load(f)


def get_measurement(letter: str) -> str | None:
    """첫 글자 → 측정량 반환."""
    return load_rules().get("measurement", {}).get(letter)


def get_modifier(letter: str) -> dict | None:
    """수식어 글자 → role 정보 반환."""
    return load_rules().get("modifiers", {}).get(letter)


def get_special_prefix(head: str) -> dict | None:
    """특수 prefix → role 정보 반환."""
    return load_rules().get("special_prefixes", {}).get(head)


def get_all_measurements() -> dict:
    """측정량 전체 표 반환 (naming_convention 시트용)."""
    return load_rules().get("measurement", {})


def get_all_modifiers() -> dict:
    """수식어 전체 표 반환 (naming_convention 시트용)."""
    return load_rules().get("modifiers", {})


def get_all_special_prefixes() -> dict:
    """특수 prefix 전체 표 반환 (naming_convention 시트용)."""
    return load_rules().get("special_prefixes", {})
```

### A-4. `mcp-server/instrument_inference/infer.py`

**파일**: `mcp-server/instrument_inference/infer.py` (신규)

핵심 유추 알고리즘. 보안 플랜 §7의 의사코드를 실제 Python 코드로 구현.

```python
"""DCS base_tag → 현장 계기 자동 유추 알고리즘."""
from __future__ import annotations
import re

from .rules import get_measurement, get_modifier, get_special_prefix


def split_tag(base_tag: str) -> tuple[str, str]:
    """'ficq-6101' → ('ficq', '6101') / 'xv-6124' → ('xv', '6124')."""
    m = re.match(r"^([a-zA-Z]+)-?(.*)", base_tag)
    if not m:
        return base_tag, ""
    return m.group(1).lower(), m.group(2)


def infer_instruments_for_base_tag(
    base_tag: str,
    data_points: list[str],
    area: str,
) -> list[dict]:
    """
    단일 base_tag에 대해 현장 계기 목록을 유추.

    Returns:
        instrument dict 리스트 (§3.1 스키마와 일치).
        매칭 실패 시 1행 포함 (confidence=low, needs_review=True).
    """
    head, loop = split_tag(base_tag)
    if not loop:
        loop = "unknown"

    dp_set = set(data_points) if data_points else set()

    # 1. 특수 prefix 우선
    sp = get_special_prefix(head)
    if sp:
        return [_build_special(head, loop, area, sp, dp_set)]

    # 2. 첫 글자 = 측정량
    first = head[0]
    meas = get_measurement(first)
    if not meas:
        return [_build_unmatched(base_tag, area, head, loop)]

    instruments = []
    has_transmitter = False
    has_controller = False

    # 3. 수식어 글자별로 계기 생성
    for letter in head[1:]:
        mod = get_modifier(letter)
        if not mod:
            continue
        if mod.get("virtual"):
            continue  # I, R은 가상

        role = mod["role"]
        inst = _build_instrument(first, meas, role, loop, area, mod, dp_set, base_tag)
        instruments.append(inst)

        if role == "transmitter":
            has_transmitter = True
        if role == "controller":
            has_controller = True

    # 4. T 글자가 명시 안 됐어도 컨트롤러가 있으면 송신기 암시
    if not has_transmitter and has_controller:
        inst = _build_implicit_transmitter(first, meas, loop, area, dp_set, base_tag)
        instruments.insert(0, inst)

    # 5. 컨트롤러 → 제어밸브 자동 생성 (auto_pair)
    if has_controller:
        inst = _build_paired_valve(first, meas, loop, area, dp_set, base_tag)
        instruments.append(inst)

    # [H-4 수정] 빈 리스트일 때 implicit transmitter 강제 추가
    # pi-XXXX/ti-XXXX 등 "측정량 + I(virtual)"만 있는 경우 instruments가 비어 있음
    if not instruments:
        inst = _build_implicit_transmitter(first, meas, loop, area, dp_set, base_tag)
        instruments.append(inst)

    # 6. from/to 채우기
    _link_signal_flow(instruments, base_tag, area)

    # 7. confidence 계산
    for inst in instruments:
        inst["confidence"] = _score_confidence(inst, dp_set)
        inst["needs_review"] = inst["confidence"] == "low"

    return instruments


def _build_instrument(
    first_letter: str, meas: str, role: str, loop: str, area: str, mod: dict, dp_set: set, parent_tag: str
) -> dict:
    role_id = _role_to_id(first_letter, role, loop)
    display = role_id.upper()
    dps = mod.get("data_points", [])
    matched_dps = [d for d in dps if d in dp_set]

    return {
        "instrument_id": role_id,
        "display_name": display,
        "parent_base_tag": parent_tag,
        "role": role,
        "loop": loop,
        "area": area or "(none)",
        "measures": meas,
        "data_points": ",".join(matched_dps) if matched_dps else "(none)",
        "from": "(none)",
        "to": "(none)",
        "description": "",
        "confidence": "medium",
        "needs_review": False,
        "inference_basis": f"{first_letter.upper()}+{mod.get('role','?')}",
        "operator_notes": "",
        "delete": False,
    }


def _build_special(head: str, loop: str, area: str, sp: dict, dp_set: set) -> list[dict]:
    role = sp["role"]
    meas = sp.get("measures")
    role_id = f"{head}-{loop}"
    display = f"{head.upper()}-{loop}"

    dp_map = {
        "shutdown-valve": [".instate0", ".instate1"],
        "interlock-relay": [".instate0", ".instate1"],
        "positioner": [".op"],
        "motor": [".run", ".fault"],
        "pump": [".run", ".flow"],
        "compressor": [],
    }
    expected_dps = dp_map.get(role, [])
    matched_dps = [d for d in expected_dps if d in dp_set]

    inst = {
        "instrument_id": role_id,
        "display_name": display,
        "parent_base_tag": f"{head}-{loop}",
        "role": role,
        "loop": loop,
        "area": area or "(none)",
        "measures": meas or "(none)",
        "data_points": ",".join(matched_dps) if matched_dps else "(none)",
        "from": "(none)",
        "to": "(none)",
        "description": "",
        "confidence": "medium",
        "needs_review": False,
        "inference_basis": f"special_prefix:{head}",
        "operator_notes": "",
        "delete": False,
    }
    return [inst]


def _build_unmatched(base_tag: str, area: str, head: str, loop: str) -> list[dict]:
    return [{
        "instrument_id": f"{head}-{loop}",
        "display_name": f"{head.upper()}-{loop}",
        "parent_base_tag": base_tag,
        "role": "equipment",
        "loop": loop,
        "area": area or "(none)",
        "measures": "(none)",
        "data_points": "(none)",
        "from": "(none)",
        "to": "(none)",
        "description": "",
        "confidence": "low",
        "needs_review": True,
        "inference_basis": "unmatched_prefix",
        "operator_notes": "",
        "delete": False,
    }]


def _build_implicit_transmitter(first: str, meas: str, loop: str, area: str, dp_set: set, parent: str) -> dict:
    role_id = f"{first}t-{loop}"
    return {
        "instrument_id": role_id,
        "display_name": f"{first.upper()}T-{loop}",
        "parent_base_tag": parent,
        "role": "transmitter",
        "loop": loop,
        "area": area or "(none)",
        "measures": meas,
        "data_points": ".pv" if ".pv" in dp_set else "(none)",
        "from": f"process/{area or 'unknown'}-{loop}-inlet",
        "to": f"tag/{parent}",
        "description": "",
        "confidence": "medium",
        "needs_review": False,
        "inference_basis": f"{first.upper()}+(implied T)",
        "operator_notes": "",
        "delete": False,
    }


def _build_paired_valve(first: str, meas: str, loop: str, area: str, dp_set: set, parent: str) -> dict:
    role_id = f"{first}cv-{loop}"
    return {
        "instrument_id": role_id,
        "display_name": f"{first.upper()}CV-{loop}",
        "parent_base_tag": parent,
        "role": "control-valve",
        "loop": loop,
        "area": area or "(none)",
        "measures": "(none)",
        "data_points": ".op" if ".op" in dp_set else "(none)",
        "from": f"tag/{first}ic-{loop}",
        "to": f"process/{area or 'unknown'}-{loop}-downstream",
        "description": "",
        "confidence": "high",
        "needs_review": False,
        "inference_basis": "C -> CV auto_pair",
        "operator_notes": "",
        "delete": False,
    }


def _role_to_id(first_letter: str, role: str, loop: str) -> str:
    mapping = {
        "transmitter": f"{first_letter}t",
        "controller": f"{first_letter}ic",
        "totalizer": f"{first_letter}q",
        "switch": f"{first_letter}s",
        "alarm": f"{first_letter}a",
        "interlock-relay": f"{first_letter}y",
        "positioner": f"{first_letter}z",
    }
    return f"{mapping.get(role, first_letter + role[:2])}-{loop}"


def _link_signal_flow(instruments: list[dict], parent_tag: str, area: str) -> None:
    """role별 from/to 기본값 채움 (§5.1)."""
    transmitters = [i for i in instruments if i["role"] == "transmitter"]
    controllers = [i for i in instruments if i["role"] == "controller"]
    valves = [i for i in instruments if i["role"] == "control-valve"]
    totalizers = [i for i in instruments if i["role"] == "totalizer"]

    for t in transmitters:
        if t["from"] == "(none)":
            t["from"] = f"process/{area or 'unknown'}-{t['loop']}-inlet"
        if t["to"] == "(none)":
            t["to"] = f"tag/{parent_tag}"

    for c in controllers:
        if c["from"] == "(none)":
            c["from"] = f"tag/{transmitters[0]['instrument_id']}" if transmitters else f"tag/{parent_tag}"
        if c["to"] == "(none)":
            c["to"] = f"tag/{valves[0]['instrument_id']}" if valves else f"tag/{parent_tag}"

    for v in valves:
        if v["from"] == "(none)":
            v["from"] = f"tag/{controllers[0]['instrument_id']}" if controllers else f"tag/{parent_tag}"
        if v["to"] == "(none)":
            v["to"] = f"process/{area or 'unknown'}-{v['loop']}-downstream"

    for tz in totalizers:
        if tz["from"] == "(none)":
            tz["from"] = f"tag/{parent_tag}"
        tz["to"] = "(none)"


def _score_confidence(inst: dict, dp_set: set) -> str:
    """data_point 일치도 + prefix 매칭으로 신뢰도 계산."""
    role = inst["role"]
    basis = inst.get("inference_basis", "")

    if basis == "unmatched_prefix":
        return "low"

    expected = {
        "transmitter": [".pv"],
        "controller": [".sp", ".op"],
        "totalizer": [".qv"],
        "switch": [".instate0"],
        "shutdown-valve": [".instate0"],
    }

    checks = expected.get(role, [])
    if not checks:
        return "medium" if basis.startswith("special") else "high"

    matched = sum(1 for c in checks if c in dp_set)
    if matched == len(checks):
        return "high"
    if matched > 0:
        return "medium"
    return "low"


def _role_to_korean_description(role: str, meas: str) -> str:
    """role + 측정량 → 한국어 설명 초안."""
    meas_ko = {
        "flow": "유량", "pressure": "압력", "temperature": "온도",
        "level": "위차", "analysis": "분석", "speed": "회전수",
        "weight": "중량", "density": "비중", "power": "전력", "moisture": "습도",
    }
    m = meas_ko.get(meas, meas)

    role_desc = {
        "transmitter": f"{m} 송신기",
        "controller": f"{m} 제어기",
        "totalizer": f"{m} 적산기",
        "switch": f"{m} 스위치",
        "alarm": f"{m} 알람",
        "control-valve": f"{m} 제어밸브",
        "shutdown-valve": "차단밸브",
        "interlock-relay": "인터록 릴레이",
        "positioner": "포지셔너",
        "motor": "모터",
        "pump": "펌프",
        "compressor": "압축기",
    }
    return role_desc.get(role, role)
```

### A-5. `mcp-server/instrument_inference/excel.py`

**파일**: `mcp-server/instrument_inference/excel.py` (신규)

3시트 Excel 생성기. openpyxl 기반.

```python
"""3시트 Excel 초안 생성기 (§3 스키마)."""
from __future__ import annotations
import os
import tempfile
from datetime import datetime

import openpyxl
from openpyxl.styles import Font, PatternFill, Alignment

from .rules import get_all_measurements, get_all_modifiers, get_all_special_prefixes


INSTRUMENT_COLS = [
    "instrument_id", "display_name", "parent_base_tag", "role", "loop",
    "area", "measures", "data_points", "from", "to", "description",
    "confidence", "needs_review", "inference_basis", "operator_notes", "delete",
]

HEADER_FILL = PatternFill(start_color="4472C4", end_color="4472C4", fill_type="solid")
HEADER_FONT = Font(bold=True, color="FFFFFF", size=11)
LOW_FILL = PatternFill(start_color="FFC7CE", end_color="FFC7CE", fill_type="solid")
MEDIUM_FILL = PatternFill(start_color="FFEB9C", end_color="FFEB9C", fill_type="solid")


def generate_excel(instruments: list[dict], unmatched: list[str], output_dir: str | None = None) -> str:
    """
    3시트 Excel 생성 후 파일 경로 반환.

    Args:
        instruments: infer.py 결과 (instrument dict 리스트)
        unmatched:   룰 미매칭 base_tag 목록
        output_dir:  출력 디렉토리 (기본: /tmp)

    Returns:
        생성된 xlsx 파일의 절대 경로
    """
    if output_dir is None:
        output_dir = tempfile.gettempdir()
    os.makedirs(output_dir, exist_ok=True)

    ts = datetime.now().strftime("%Y%m%d_%H%M%S")
    filepath = os.path.join(output_dir, f"instruments_draft_{ts}.xlsx")

    wb = openpyxl.Workbook()

    # ── 시트 1: instruments ──
    _build_instruments_sheet(wb, instruments)

    # ── 시트 2: naming_convention ──
    _build_naming_convention_sheet(wb)

    # ── 시트 3: unmatched_tags ──
    _build_unmatched_sheet(wb, unmatched)

    wb.save(filepath)
    return filepath


def _build_instruments_sheet(wb: openpyxl.Workbook, instruments: list[dict]) -> None:
    ws = wb.active
    ws.title = "instruments"

    # 헤더
    for col_idx, col_name in enumerate(INSTRUMENT_COLS, 1):
        cell = ws.cell(row=1, column=col_idx, value=col_name)
        cell.font = HEADER_FONT
        cell.fill = HEADER_FILL
        cell.alignment = Alignment(horizontal="center", wrap_text=True)

    # 데이터 행
    for row_idx, inst in enumerate(instruments, 2):
        for col_idx, col_name in enumerate(INSTRUMENT_COLS, 1):
            val = inst.get(col_name, "")
            cell = ws.cell(row=row_idx, column=col_idx, value=val)

            # confidence별 색상
            if col_name == "confidence":
                if val == "low":
                    cell.fill = LOW_FILL
                elif val == "medium":
                    cell.fill = MEDIUM_FILL

    # 컬럼 너비 자동 조정
    for col_idx in range(1, len(INSTRUMENT_COLS) + 1):
        max_len = 0
        for row in range(1, ws.max_row + 1):
            val = str(ws.cell(row=row, column=col_idx).value or "")
            max_len = max(max_len, len(val))
        ws.column_dimensions[openpyxl.utils.get_column_letter(col_idx)].width = min(max_len + 4, 40)


def _build_naming_convention_sheet(wb: openpyxl.Workbook) -> None:
    ws = wb.create_sheet("naming_convention")

    # 측정량 표
    ws.cell(row=1, column=1, value="측정량 (첫 글자)").font = HEADER_FONT
    ws.cell(row=1, column=2, value="의미").font = HEADER_FONT
    meas = get_all_measurements()
    for i, (letter, meaning) in enumerate(meas.items(), 2):
        ws.cell(row=i, column=1, value=letter)
        ws.cell(row=i, column=2, value=meaning)

    # 수식어 표 (D열부터)
    ws.cell(row=1, column=4, value="수식어 (두 번째 이후)").font = HEADER_FONT
    ws.cell(row=1, column=5, value="role").font = HEADER_FONT
    ws.cell(row=1, column=6, value="가상").font = HEADER_FONT
    mods = get_all_modifiers()
    for i, (letter, info) in enumerate(mods.items(), 2):
        ws.cell(row=i, column=4, value=letter)
        ws.cell(row=i, column=5, value=info.get("role", ""))
        ws.cell(row=i, column=6, value=info.get("virtual", False))

    # 특수 prefix 표 (G열부터)
    ws.cell(row=1, column=7, value="특수 prefix").font = HEADER_FONT
    ws.cell(row=1, column=8, value="role").font = HEADER_FONT
    sps = get_all_special_prefixes()
    for i, (prefix, info) in enumerate(sps.items(), 2):
        ws.cell(row=i, column=7, value=prefix)
        ws.cell(row=i, column=8, value=info.get("role", ""))


def _build_unmatched_sheet(wb: openpyxl.Workbook, unmatched: list[str]) -> None:
    ws = wb.create_sheet("unmatched_tags")

    ws.cell(row=1, column=1, value="base_tag").font = HEADER_FONT
    ws.cell(row=1, column=2, value="area").font = HEADER_FONT
    ws.cell(row=1, column=3, value="action").font = HEADER_FONT
    ws.cell(row=1, column=4, value="operator_notes").font = HEADER_FONT

    for i, tag in enumerate(unmatched, 2):
        ws.cell(row=i, column=1, value=tag)
        ws.cell(row=i, column=2, value="(none)")
        ws.cell(row=i, column=3, value="운영자가 instruments 시트에 행 추가 필요")
        ws.cell(row=i, column=4, value="")
```

### A-6. `mcp-server/server.py` — 도구 등록

**파일**: `mcp-server/server.py` (수정)

기존 `parse_document` 도구 다음에 신규 도구 추가.

```python
# ── Field Instrument Inference ──────────────────────────────────────────────

@mcp.tool()
def infer_field_instruments(
    use_llm: bool = False,
    seed_doc_id: str | None = None,
) -> str:
    """
    v_tag_summary의 모든 base_tag를 대상으로 현장 계기를 자동 유추하여 Excel 초안 생성.

    Args:
        use_llm:      LLM으로 description 한국어 초안 보강 (기본 False, Phase C에서 활성화)
        seed_doc_id:  기존 Excel doc_id 지정 시 수정사항 보존 + 신규만 추가

    Returns:
        JSON: { success, doc_path, instrument_count, unmatched_count, message }
    """
    import json

    from instrument_inference.rules import load_rules
    from instrument_inference.infer import infer_instruments_for_base_tag
    from instrument_inference.excel import generate_excel

    # [H-3 수정] v_tag_summary에서 base_tag별 dp 집합을 SQL로 한 번에 추출
    # pv/sp/op/instate0/1/2 NULL 여부로 실제 data_point 유무 판정
    def _fetch_tags():
        import psycopg
        conn = psycopg.connect(DB_CONNECTION_STRING, connect_timeout=DB_TIMEOUT)
        try:
            with conn.cursor() as cur:
                cur.execute("""
                    SELECT
                        base_tag,
                        area,
                        CASE WHEN pv IS NOT NULL THEN '.pv' END,
                        CASE WHEN sp IS NOT NULL THEN '.sp' END,
                        CASE WHEN op IS NOT NULL THEN '.op' END,
                        CASE WHEN instate0 IS NOT NULL THEN '.instate0' END,
                        CASE WHEN instate1 IS NOT NULL THEN '.instate1' END,
                        CASE WHEN instate2 IS NOT NULL THEN '.instate2' END
                    FROM v_tag_summary
                    ORDER BY base_tag
                """)
                return cur.fetchall()
        finally:
            conn.close()

    # [L-3 수정] .qv는 v_tag_summary에 없으므로 별도 조회
    def _fetch_qv_tags():
        import psycopg
        conn = psycopg.connect(DB_CONNECTION_STRING, connect_timeout=DB_TIMEOUT)
        try:
            with conn.cursor() as cur:
                cur.execute("""
                    SELECT DISTINCT SUBSTRING(tagname FROM '(.*)\\.qv') AS base_tag
                    FROM realtime_table
                    WHERE tagname LIKE '%.qv'
                """)
                return set(r[0] for r in cur.fetchall())
        finally:
            conn.close()

    tags = _fetch_tags()
    # [M-1 수정] asyncio.to_thread 데드 코드 삭제 — sync def이므로 blocking 호출

    qv_tags = _fetch_qv_tags()

    all_instruments = []
    unmatched_tags = []

    for row in tags:
        base_tag = row[0]
        area = row[5]
        if not base_tag:
            continue

        # [H-3 수정] 실제 DB 컬럼 NULL 여부로 dp 집합 구성
        dp = [d for d in row[2:8] if d is not None]
        if base_tag in qv_tags:
            dp.append(".qv")

        try:
            result = infer_instruments_for_base_tag(base_tag, dp, area or "")
            # [H-4 수정] 빈 리스트 가드
            if not result:
                unmatched_tags.append(base_tag)
                continue
            if result[0].get("needs_review") and result[0].get("inference_basis") == "unmatched_prefix":
                unmatched_tags.append(base_tag)
            else:
                all_instruments.extend(result)
        except Exception as e:
            logging.warning(f"[infer] {base_tag} 처리 실패: {e}")
            unmatched_tags.append(base_tag)

    # [M-5 수정] instrument_id 기준 dedup
    seen_ids: dict[str, dict] = {}
    deduped_instruments = []
    for inst in all_instruments:
        iid = inst["instrument_id"]
        if iid in seen_ids:
            # 기존 항목의 inference_basis에 출처 결합
            existing = seen_ids[iid]
            existing["inference_basis"] = f"{existing['inference_basis']}; also from {inst.get('parent_base_tag', '?')}"
        else:
            seen_ids[iid] = inst
            deduped_instruments.append(inst)
    all_instruments = deduped_instruments

    # [M-3 수정] seed_doc_id 기반 머지 로직
    if seed_doc_id:
        all_instruments = _merge_with_seed(seed_doc_id, all_instruments)

    # Excel 생성
    xlsx_path = generate_excel(all_instruments, unmatched_tags)

    # LLM 보강 (Phase C — 아직 stub)
    if use_llm:
        xlsx_path = _enrich_descriptions_llm(xlsx_path)

    return json.dumps({
        "success": True,
        "doc_path": xlsx_path,
        "instrument_count": len(all_instruments),
        "unmatched_count": len(unmatched_tags),
        "message": f"{len(all_instruments)}개 계기 유추 완료 ({len(unmatched_tags)}개 미매칭)",
    })


def _merge_with_seed(seed_doc_id: str, new_instruments: list[dict]) -> list[dict]:
    """[M-3 수정] seed_doc_id 기반 머지 — 운영자 수정사항 보존 + 신규만 추가.

    seed_doc_id는 kb_documents.id(UUID)이므로 C# 측에서 Excel 파일을 읽어
    MCP 도구에 전달해야 함. 현재는 MCP 도구 내에서 직접 처리하지 않고,
    C# KbController에서 처리하도록 설계.
    """
    # 실제 구현은 KbController.cs에서 수행 (§B-1 참조)
    # 여기서는 placeholder — Phase A에서는 seed_doc_id 미지원
    return new_instruments


def _enrich_descriptions_llm(xlsx_path: str) -> str:
    """LLM으로 description 한국어 초안 보강 (Phase C stub)."""
    # [M-2 수정] Phase C 구현 전까지 no-op. UI는 disabled로 처리.
    return xlsx_path
```

**중요**: `infer_field_instruments`는 sync def로 등록한다. 내부에서 DB 조회는 blocking 호출로 처리한다. 이는 기존 `ask_iiot_llm` 도구와 동일한 패턴이다 (server.py:476 — sync def, 내부에서 `_llm().chat.completions.create()` blocking 호출).

**[H-3 수정 노트]**: `v_tag_summary` 뷰는 이미 `pv`, `sp`, `op`, `instate0`, `instate1`, `instate2` 컬럼을 노출하므로 NULL 여부로 dp 유무를 정확히 판정할 수 있다. `.qv`는 별도 `realtime_table` 조회 필요 (L-3).

---

## Phase B — API + UI (C# + JS)

### B-1. `KbController.cs` — `/api/kb/instruments/infer` 엔드포인트 추가

**파일**: `src/Web/Controllers/KbController.cs` (수정)

**[H-1 수정]** 생성자에 `McpClient` 주입 추가:

```csharp
public KbController(
    ExperionDbContext db,
    KbStorageService storage,
    KbQdrantClient qdrant,
    IKbAuthService auth,
    McpClient mcp,                    // ← [H-1 수정] 신규
    ILogger<KbController> logger)
{
    _db = db;
    _storage = storage;
    _qdrant = qdrant;
    _auth = auth;
    _mcp = mcp;                       // ← [H-1 수정] 신규 필드
    _logger = logger;
}
```

필드 선언도 추가:
```csharp
private readonly McpClient _mcp;      // ← [H-1 수정] 신규
```

`Program.cs:98`에 `AddSingleton<McpClient>()` 이미 등록되어 있어 별도 설정 불필요.

---

기존 `PurgeDisabled` 엔드포인트 다음에 추가:

```csharp
public sealed record InferInstrumentsRequest(bool UseLlm = false, string? SeedDocId = null);

[HttpPost("instruments/infer")]
public async Task<IActionResult> InferInstruments([FromBody] InferInstrumentsRequest? req, CancellationToken ct)
{
    if (!await RequireAdminAsync(ct))
        return Unauthorized(new { success = false, error = "unauthorized" });

    try
    {
        var args = new Dictionary<string, object>
        {
            ["use_llm"] = req?.UseLlm ?? false,
            ["seed_doc_id"] = req?.SeedDocId ?? ""
        };

        var raw = await _mcp.CallToolAsync("infer_field_instruments", args, ct);
        if (string.IsNullOrWhiteSpace(raw))
            return BadRequest(new { success = false, error = "MCP 응답 없음" });

        // [H-2 수정] McpClient.CallToolAsync는 실패 시 평문 문자열 반환
        // → KbIngestWorker.ParseAsync(cs:202-233) 패턴 그대로 복제
        try
        {
            using var jdoc = System.Text.Json.JsonDocument.Parse(raw);
            if (jdoc.RootElement.TryGetProperty("success", out var s) && s.ValueKind == System.Text.Json.JsonValueKind.False)
            {
                var err = jdoc.RootElement.TryGetProperty("message", out var e) ? e.GetString() : "unknown";
                return BadRequest(new { success = false, error = err });
            }

            var docPath = jdoc.RootElement.GetProperty("doc_path").GetString() ?? "";
            var instrumentCount = jdoc.RootElement.GetProperty("instrument_count").GetInt32();
            var unmatchedCount = jdoc.RootElement.GetProperty("unmatched_count").GetInt32();

            // Excel 파일을 KB 저장소로 복사
            var fileName = System.IO.Path.GetFileName(docPath);
            await using var stream = new FileStream(docPath, FileMode.Open, FileAccess.Read, FileShare.Read);
            var stored = await _storage.SaveAsync(stream, fileName, ct);

            // [L-2 수정] /tmp 임시 파일 정리
            try { System.IO.File.Delete(docPath); } catch { }

            // kb_documents + kb_ingest_jobs INSERT (기존 Upload와 동일한 흐름)
            var doc = new KbDocument
            {
                Id = Guid.NewGuid(),
                CollectionKey = "system_instrument",
                Title = $"Field Instruments Draft {DateTime.UtcNow:yyyyMMdd}",
                OriginalPath = stored.RelativePath,
                FileSha256 = stored.Sha256,
                FileSize = stored.Size,
                MimeType = "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
                Tags = Array.Empty<string>(),
                Status = "pending",
                ChunkCount = 0,
                UploadedAt = DateTime.UtcNow,
                UploadedBy = HttpContext.Connection.RemoteIpAddress?.ToString()
            };
            _db.KbDocuments.Add(doc);

            _db.KbIngestJobs.Add(new KbIngestJob
            {
                DocId = doc.Id,
                Stage = "parse",
                EnqueuedAt = DateTime.UtcNow
            });
            await _db.SaveChangesAsync(ct);

            _logger.LogInformation("[Kb][Infer] {Id} {Count} instruments, {Unmatched} unmatched", doc.Id, instrumentCount, unmatchedCount);

            return Ok(new
            {
                success = true,
                docId = doc.Id,
                status = doc.Status,
                instrumentCount,
                unmatchedCount,
                message = jdoc.RootElement.TryGetProperty("message", out var m) ? m.GetString() : ""
            });
        }
        catch (System.Text.Json.JsonException)
        {
            // [H-2 수정] 평문 응답("도구 호출 실패: ...")일 경우
            _logger.LogWarning("[Kb][Infer] MCP 평문 응답: {Raw}", raw);
            return BadRequest(new { success = false, error = "MCP 도구 실패: " + raw.TrimEnd('\r', '\n', ' ') });
        }
    }
    catch (Exception ex)
    {
        _logger.LogError(ex, "[Kb][Infer] 실패");
        return BadRequest(new { success = false, error = ex.Message });
    }
}
```

**검증**:
- `[H-1]` `McpClient mcp` 생성자 주입 + `private readonly McpClient _mcp;` 필드 추가 ✓
- `[H-2]` `try/catch(JsonException)` 패턴 복제 + 평문 응답 시 에러 메시지 전달 ✓
- `[L-2]` `FileStream` 블록 종료 직후 `File.Delete(docPath)` 추가 ✓
- `RequireAdminAsync` 사용 ✓ (기존 패턴)
- `Ok(new { success = true, docId = ... })` — 명시적 camelCase ✓ (CODING_CONVENTIONS.md)
- `KbDocument`/`KbIngestJob` 생성 — 기존 `Upload`와 동일한 패턴 ✓

### B-1-2. seed_doc_id 머지 로직 ([M-3 수정])

**파일**: `src/Web/Controllers/KbController.cs` (수정)

`InferInstruments` 엔드포인트에서 `seed_doc_id`가 있으면 기존 Excel을 로드하여 머지:

```csharp
// InferInstruments 메서드 내에서 args 생성 전에 추가:
string? seedId = req?.SeedDocId;
if (!string.IsNullOrWhiteSpace(seedId))
{
    // 기존 문서 조회
    var seedDoc = await _db.KbDocuments
        .FirstOrDefaultAsync(d => d.Id == Guid.Parse(seedId), ct);
    if (seedDoc != null)
    {
        var seedPath = _storage.Resolve(seedDoc.OriginalPath);
        if (System.IO.File.Exists(seedPath))
        {
            // args에 seed_excel_path 추가 (MCP 도구가 로드하여 머지)
            args["seed_excel_path"] = seedPath;
        }
    }
}
```

대응 MCP 도구 측 (`server.py`의 `infer_field_instruments`)에서 `seed_excel_path` 파라미터로 기존 Excel을 openpyxl로 로드:

```python
# server.py infer_field_instruments 내부 — [M-3 수정]
seed_excel_path = None  # 기본값

# args에서 가져오기 (C# 측에서 설정)
# seed_excel_path는 MCP arguments로 전달됨

if seed_excel_path and os.path.exists(seed_excel_path):
    existing = _load_seed_instruments(seed_excel_path)
    # instrument_id 기준 머지:
    #   - 기존 행 + delete=FALSE → 보존 (운영자 수정사항 유지)
    #   - 신규 instrument_id → 추가
    #   - 기존 행 + delete=TRUE → 제외
    all_instruments = _merge_instruments(existing, all_instruments)


def _load_seed_instruments(seed_path: str) -> list[dict]:
    """기존 Excel에서 instruments 시트 로드."""
    from openpyxl import load_workbook
    wb = load_workbook(seed_path, read_only=True, data_only=True)
    ws = wb["instruments"]
    rows = list(ws.iter_rows(values_only=True))
    if not rows:
        return []
    headers = [str(c) for c in rows[0]]
    result = []
    for r in rows[1:]:
        inst = dict(zip(headers, r))
        # delete=TRUE인 행은 제외
        if inst.get("delete") in (True, "TRUE", "True", "YES", "Yes", "Y", "y"):
            continue
        result.append(inst)
    return result


def _merge_instruments(existing: list[dict], new_instruments: list[dict]) -> list[dict]:
    """운영자 수정사항 보존 + 신규만 추가."""
    existing_ids = {i["instrument_id"] for i in existing}
    merged = list(existing)  # 기존 행 보존
    for inst in new_instruments:
        if inst["instrument_id"] not in existing_ids:
            merged.append(inst)
    return merged
```

### B-2. `index.html` — 14번 탭에 Field Instrument 섹션 추가

**파일**: `src/Web/wwwroot/index.html` (수정)

`#kb-main` 내부, `.kb-topbar` 다음에 Field Instrument 카드 추가:

```html
<!-- Field Instrument 초안 생성 -->
<div class="card kb-infer-card" style="margin-top:12px">
  <div class="card-cap">🔧 Field Instrument 자동 유추</div>
  <div class="kb-infer-body">
    <p class="kb-infer-desc">DCS 태그 ficq-6101 → FT-6101, FIC-6101, FCV-6101 등 현장 계기 자동 유추 후 Excel 초안 생성.</p>
    <div class="kb-infer-options">
      <!-- [M-2 수정] Phase C 구현 전까지 disabled -->
      <label class="kb-toggle" title="Phase C에서 활성화 예정입니다">
        <input type="checkbox" id="kb-infer-llm" disabled/>
        <span>LLM description 보강 (Phase C에서 활성화)</span>
      </label>
    </div>
    <div class="btn-row">
      <button class="btn-a btn-sm" id="kb-infer-btn" onclick="kbInferStart()">⚡ 초안 생성</button>
      <span id="kb-infer-msg" class="kb-msg"></span>
    </div>
    <div id="kb-infer-result" class="kb-infer-result hidden"></div>
  </div>
</div>
```

### B-3. `app.js` — infer 시작/결과 처리

**파일**: `src/Web/wwwroot/js/app.js` (수정)

`kbStartPoll`/`kbStopPoll` 함수 다음에 추가:

```javascript
// ── Field Instrument Inference ──────────────────────────────
async function kbInferStart() {
  const btn = document.getElementById('kb-infer-btn');
  const msg = document.getElementById('kb-infer-msg');
  const result = document.getElementById('kb-infer-result');
  const useLlm = document.getElementById('kb-infer-llm').checked;

  btn.disabled = true;
  btn.textContent = '⏳ 생성 중...';
  msg.textContent = '계기 유추 및 Excel 생성 중 (수십 초 소요)';
  result.classList.add('hidden');

  try {
    const r = await kbFetch('POST', '/api/kb/instruments/infer', {
      useLlm,
      seedDocId: ''
    });

    if (!r.ok || !r.data || !r.data.success) {
      msg.textContent = '❌ ' + (r.data && r.data.error ? r.data.error : 'HTTP ' + r.status);
      return;
    }

    const d = r.data;
    msg.textContent = '✅ 생성 완료. 인덱싱 진행 중...';
    result.classList.remove('hidden');
    result.innerHTML = `
      <div class="kb-infer-stats">
        <span class="kb-stat">유추: ${d.instrumentCount}개 계기</span>
        <span class="kb-stat">미매칭: ${d.unmatchedCount}개</span>
      </div>
      <p>${d.message || ''}</p>
      <div class="btn-row">
        <button class="btn-b btn-sm" onclick="kbDownload('${d.docId}')">⬇ Excel 다운로드</button>
        <button class="btn-b btn-sm" onclick="kbRefresh()">🔄 목록 새로고침</button>
      </div>
    `;

    // 목록 새로고침 (인덱싱 상태 폴링은 기존 kbStartPoll이 처리)
    setTimeout(() => kbRefresh(), 2000);
  } catch (e) {
    msg.textContent = '❌ ' + e.message;
  } finally {
    btn.disabled = false;
    btn.textContent = '⚡ 초안 생성';
  }
}
```

### B-4. `style.css` — 관련 스타일

**파일**: `src/Web/wwwroot/css/style.css` (수정)

기존 `.kb-*` 스타일 다음에 추가:

```css
/* Field Instrument Inference */
.kb-infer-card { border-left: 3px solid #4472C4; }
.kb-infer-body { padding: 8px 0; }
.kb-infer-desc { color: #666; font-size: 0.9em; margin: 8px 0; }
.kb-infer-options { margin: 8px 0; }
.kb-toggle { display: flex; align-items: center; gap: 6px; cursor: pointer; font-size: 0.9em; }
.kb-toggle input { margin: 0; }
.kb-infer-result { margin-top: 12px; padding: 12px; background: #f0f7ff; border-radius: 6px; }
.kb-infer-stats { display: flex; gap: 16px; margin-bottom: 8px; }
.kb-stat { font-weight: 600; color: #333; }
.kb-stat::before { content: ''; display: inline-block; width: 8px; height: 8px; border-radius: 50%; background: #4472C4; margin-right: 4px; }
```

---

## Phase C — LLM 보강 (옵션, 이후 구현)

`infer.py`에 `enrich_with_llm(rows, batch=50)` 함수 추가.

```python
def enrich_with_llm(instruments: list[dict], batch_size: int = 50) -> list[dict]:
    """description이 비어 있거나 confidence=low인 행에 LLM으로 한국어 설명 보강."""
    from openai import OpenAI

    client = OpenAI(base_url=VLLM_BASE_URL, api_key="dummy")
    model = VLLM_MODEL

    # plant_context.md 로드
    context_path = os.path.join(os.path.dirname(__file__), "..", "..", "prompts", "plant_context.md")
    context = ""
    if os.path.exists(context_path):
        with open(context_path, "r", encoding="utf-8") as f:
            context = f.read()

    # 대상 필터: description 비어 있거나 confidence=low
    targets = [i for i in instruments if not i.get("description") or i.get("confidence") == "low"]

    # 캐시: (parent_base_tag, role) → description
    cache: dict[tuple[str, str], str] = {}

    for idx in range(0, len(targets), batch_size):
        batch = targets[idx:idx + batch_size]
        batch_text = "\n".join(
            f"- {t['display_name']} ({t['role']}): loop={t['loop']}, area={t['area']}, "
            f"measures={t['measures']}, from={t['from']}, to={t['to']}"
            for t in batch
        )

        key = (batch_text, context)
        if key in cache:
            continue

        try:
            resp = client.chat.completions.create(
                model=model,
                messages=[
                    {"role": "system", "content": (
                        f"당신은 플랜트 계기 엔지니어입니다. 아래 플랜트 정보를 참고하세요:\n{context}\n\n"
                        "해당 계기들의 한국어 설명(각각 1줄, 80자 이내)을 생성하세요. "
                        "확실치 않으면 비워두세요. 형식: display_name|설명"
                    )},
                    {"role": "user", "content": batch_text},
                ],
                max_tokens=1024,
                temperature=0.1,
            )
            text = resp.choices[0].message.content or ""
            for line in text.strip().split("\n"):
                if "|" in line:
                    name, desc = line.split("|", 1)
                    cache[(name.strip(), "")] = desc.strip()
        except Exception as e:
            logging.warning(f"[infer] LLM 보강 실패: {e}")

    # 캐시 적용
    for inst in targets:
        key = (inst["display_name"], "")
        if key in cache and cache[key]:
            inst["description"] = cache[key]

    return instruments
```

---

## Phase D — 검색 소비 강화

### D-1. `search_kb` 가중치 조정

**파일**: `mcp-server/server.py` (수정)

`_search_kb_raw` 함수에서 `system_instrument` 컬렉션에 가중치 부여. 현재 `_recency_factor`와 유사하게 컬렉션별 boost를 추가:

```python
def _collection_boost(collection_key: str) -> float:
    """system_instrument 컬렉션 우선 가중."""
    if collection_key == "system_instrument":
        return 1.15
    return 1.0
```

`_search_kb_raw` 내 score 계산 부분 수정:
```python
base_score = h.get("score", 0.0)
recency = _recency_factor(uploaded_at) if boost_recent else 1.0
collection_b = _collection_boost(p.get("collection_key", ""))
results.append({
    "score": base_score * recency * collection_b,
    ...
})
```

### D-2. `ToolGuideKo` 수정

**파일**: `src/Web/Controllers/OllamaController.cs` (수정)

`ToolGuideKo`의 `search_kb` 설명에 한 줄 추가:

```
- search_kb: 사용자 지식 베이스 검색. FT/FCV 등 현장 계기명 질의는 system_instrument 컬렉션 우선 검색됨.
```

---

## 구현 순서 및 예상 소요 시간

| 순서 | 작업 | 파일 | 예상 시간 |
|------|------|------|-----------|
| 1 | pyyaml 의존성 + hatch build 수정 | `pyproject.toml` | 5분 |
| 2 | YAML 룰 파일 작성 | `prompts/instrument_inference.yaml` | 10분 |
| 3 | rules.py, infer.py, excel.py 작성 | `instrument_inference/` | 1시간 |
| 4 | MCP 도구 등록 | `server.py` | 30분 |
| 5 | Controller 엔드포인트 | `KbController.cs` | 30분 |
| 6 | UI HTML + JS + CSS | `index.html`, `app.js`, `style.css` | 1시간 |
| 7 | 테스트 + 검증 | — | 30분 |
| 8 | LLM 보강 (Phase C, 옵션) | `infer.py` | 30분 |
| 9 | 검색 가중치 + ToolGuideKo (Phase D) | `server.py`, `OllamaController.cs` | 20분 |

**총: 약 4시간 (Phase C/D 포함)**

---

## 테스트 시나리오

### T-1. 기본 유추 테스트

```bash
# ficq-6101 → FT-6101, FIC-6101, FCV-6101, FQ-6101
python -c "
from instrument_inference.infer import infer_instruments_for_base_tag
result = infer_instruments_for_base_tag('ficq-6101', ['.pv', '.sp', '.op', '.qv'], 'A')
for r in result:
    print(f\"{r['instrument_id']:15} {r['role']:20} {r['confidence']:8} {r['inference_basis']}\")
"
```

**기대 출력**:
```
ft-6101         transmitter          high     F+(implied T)
fic-6101        controller           high     F+C
fcv-6101        control-valve        high     C -> CV auto_pair
fq-6101         totalizer            high     F+Q
```

### T-2. 특수 prefix 테스트

```bash
python -c "
from instrument_inference.infer import infer_instruments_for_base_tag
result = infer_instruments_for_base_tag('xv-6124', ['.instate0', '.instate1'], 'A')
for r in result:
    print(f\"{r['instrument_id']:15} {r['role']:20} {r['confidence']:8} {r['needs_review']}\")
"
```

**기대 출력**:
```
xv-6124         shutdown-valve       high     False
```
> **[M-4 수정]**: `_build_special`이 초기 confidence="medium"을 설정하지만, 함수 끝의 `_score_confidence`가 `shutdown-valve` role + `.instate0` dp 매칭 → "high"로 덮어씀. 기대 출력을 "high"로 정정.

### T-3. 비표준 prefix 테스트

```bash
python -c "
from instrument_inference.infer import infer_instruments_for_base_tag
result = infer_instruments_for_base_tag('zz-9999', [], '')
for r in result:
    print(f\"{r['instrument_id']:15} {r['role']:20} {r['confidence']:8} {r['needs_review']}\")
"
```

**기대 출력**:
```
zz-9999         equipment            low      True
```

### T-4. Excel 생성 테스트

```bash
python -c "
from instrument_inference.infer import infer_instruments_for_base_tag
from instrument_inference.excel import generate_excel

instruments = []
for tag in ['ficq-6101', 'pic-6201', 'tic-6301', 'xv-6124', 'p-6102']:
    instruments.extend(infer_instruments_for_base_tag(tag, ['.pv', '.sp', '.op'], 'A'))

path = generate_excel(instruments, ['zz-9999'])
print(f'생성: {path}')
"
```

### T-5. 통합 테스트 (API 레벨)

1. 14번 탭 로그인
2. "⚡ 초안 생성" 클릭
3. 진행 메시지 확인 → "✅ 생성 완료"
4. 문서 목록에 새 행 등장 (status: pending → parsing → embedding → indexed)
5. ⬇ 클릭 → Excel 다운로드 → 3시트 확인
6. 🔍 클릭 → Qdrant 청크 확인
7. 채팅에서 "FT-6101 어디 붙어있어?" 질문 → KB hit 확인

---

## 재실행 정책 (seed_doc_id 기반 머지)

보안 플랜 §9에서 언급된 재실행 정책 구현:

1. `seed_doc_id`가 있으면 기존 Excel을 openpyxl로 로드
2. 기존 `instrument_id` 행을 메모리에 캐싱
3. 신규 유추 결과와 비교:
   - 기존 행 + `delete=FALSE` → 보존 (운영자 수정사항 유지)
   - 신규 `instrument_id` → 추가
   - 기존 행 + `delete=TRUE` → 제외
4. 합친 결과로 새 Excel 생성

---

## vault 마이그레이션 준비

보안 플랜 §11의 변환 매핑을 참고하여, Excel 열명이 vault 노트 frontmatter와 1:1 대응되도록 설계됨. 향후 옵시디언 vault 구현 시 이 Excel을 그대로 시드로 사용.

---

## 운영 가정 ([L-1 수정])

- **MCP 서버와 C# 웹앱은 동일 호스트 파일시스템을 공유한다.** `infer_field_instruments` 도구가 `/tmp`에 Excel을 생성하면 C# `KbController`가 동일 경로를 `FileStream`으로 읽는다.
- 향후 컨테이너/Docker 분리 배포 시: MCP 응답에 base64 인코딩 또는 공유 볼륨 경로로 전환 필요.
- **mcp-server Python 가상환경에 `pyyaml` 패키지가 설치되어 있어야 한다.** (`uv pip install pyyaml`)

---

## 요약

| 항목 | 내용 |
|------|------|
| 신규 인프라 | 없음 (기존 MCP + KB + Qdrant 재사용) |
| 신규 의존성 | pyyaml 1개 |
| 신규 파일 | 5개 (YAML 1, Python 3, __init__.py 1) |
| 수정 파일 | 6개 (server.py, KbController.cs, pyproject.toml, index.html, app.js, style.css) |
| 예상 소요 | 4.5시간 (검수 반영으로 +30분) |
| 위험도 | 낮음 (기존 인프라 100% 재사용, 휴먼-인-더-루프 검증) |

---

## 검수 반영 후 변경 사항 요약

| 섹션 | 변경 내용 |
|------|-----------|
| A-4 (infer.py) | H-4: 빈 리스트 가드 추가 (pi-XXXX 등 virtual-only 태그 처리) |
| A-6 (server.py) | H-3: `v_tag_summary`에서 dp 집합 SQL 추출 + `.qv` 별도 조회 |
| A-6 (server.py) | M-1: `asyncio.to_thread` 데드 코드 삭제 |
| A-6 (server.py) | M-3: seed_doc_id 머지 로직 + `_merge_with_seed` placeholder |
| A-6 (server.py) | M-5: `instrument_id` 기준 dedup 로직 |
| B-1 (KbController) | H-1: `McpClient` 생성자 주입 + 필드 추가 |
| B-1 (KbController) | H-2: `try/catch(JsonException)` + 평문 응답 처리 |
| B-1 (KbController) | L-2: `/tmp` 파일 삭제 코드 |
| B-1-2 (KbController) | M-3: seed_doc_id 머지 실제 코드 (C# 측 + Python 측) |
| B-2 (index.html) | M-2: LLM 토글 disabled + 설명 변경 |
| 테스트 T-2 | M-4: 기대 출력 "medium" → "high" 정정 |
| 신규 섹션 | 운영 가정 (L-1) |