diff --git a/pyproject.toml b/pyproject.toml
index fa6c6f2..572d328 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -15,6 +15,7 @@ dependencies = [
   "discord.py>=2.3,<3.0",   # Phase 1 frontend transport
   "cryptography>=42.0",      # EncryptedSecrets at-rest encryption
   "sqlalchemy>=2.0",         # generic DB explorer (one adapter, many engines)
+  "PyYAML>=6.0",             # OKF bundle frontmatter serialization
 ]
 
 [project.optional-dependencies]
diff --git a/src/lang2sql/adapters/storage/okf_bundle.py b/src/lang2sql/adapters/storage/okf_bundle.py
new file mode 100644
index 0000000..1d3992e
--- /dev/null
+++ b/src/lang2sql/adapters/storage/okf_bundle.py
@@ -0,0 +1,183 @@
+"""OkfBundle — OKF(Open Knowledge Format) 기반 지식 번들 어댑터.
+
+KV 캐시(SqliteStore)와 양방향 sync:
+- export: KV → 스코프별 .md 파일 (Git 영속, 사람이 읽을 수 있는 형태)
+- import_: .md 파일 → KV (번들에서 런타임 캐시 복원)
+
+디렉토리 구조 (OKF SPEC §3):
+    <base_dir>/
+    ├── guild/
+    │   ├── index.md
+    │   ├── metrics/active_user.md
+    │   ├── tables/orders.md
+    │   ├── rules/exclude_cancelled.md
+    │   ├── dimensions/customer_tier.md
+    │   └── misc/<term>.md       # kind 미지정
+    └── channel:<ch_id>/
+        ├── index.md
+        └── metrics/active_user.md
+
+각 .md 파일 형식 (OKF SPEC §4):
+    ---
+    type: Metric
+    title: active_user
+    description: "30일 내 로그인한 users"
+    tags: [growth, retention]
+    applies_to: users
+    synonyms: [활성화고객]
+    layer: guild
+    entity: ""
+    inferred: false
+    timestamp: 2026-07-18T...
+    ---
+
+    (markdown body — definition 반복 또는 추가 설명)
+"""
+
+from __future__ import annotations
+
+import re
+from datetime import datetime, timezone
+from pathlib import Path
+from typing import TYPE_CHECKING
+
+import yaml
+
+from ...tools.semantic_federation import (
+    FedEntry,
+    _KV_PREFIX,
+    _kv_key,
+)
+
+if TYPE_CHECKING:
+    from .sqlite_store import SqliteStore
+
+_KIND_FOLDER: dict[str, str] = {
+    "metric": "metrics",
+    "table": "tables",
+    "rule": "rules",
+    "dimension": "dimensions",
+}
+_RESERVED = {"index.md", "log.md"}
+
+
+class OkfBundle:
+    """KV ↔ OKF .md 파일 양방향 sync 어댑터."""
+
+    def __init__(self, base_dir: str) -> None:
+        self.base_dir = Path(base_dir)
+
+    # ------------------------------------------------------------------
+    # Public API
+    # ------------------------------------------------------------------
+
+    def export(self, store: "SqliteStore", kv_scope: str) -> int:
+        """KV에서 모든 FedEntry를 읽어 .md 파일로 저장. 저장된 파일 수 반환."""
+        raw = store.kv_list_prefix(kv_scope, _KV_PREFIX + ":")
+        count = 0
+        for _key, val in raw:
+            try:
+                entry = FedEntry.from_json(val)
+                path = self._concept_path(entry)
+            except (ValueError, KeyError):
+                continue
+            path.parent.mkdir(parents=True, exist_ok=True)
+            path.write_text(_entry_to_md(entry), encoding="utf-8")
+            count += 1
+        return count
+
+    def import_(self, store: "SqliteStore", kv_scope: str) -> int:
+        """번들의 .md 파일을 읽어 KV로 복원. 로드된 항목 수 반환."""
+        count = 0
+        for md_file in self.base_dir.rglob("*.md"):
+            if md_file.name in _RESERVED:
+                continue
+            entry = _md_to_entry(md_file)
+            if entry is None:
+                continue
+            key = _kv_key(entry.term, entry.layer, entry.entity)
+            store.kv_set(kv_scope, key, entry.to_json())
+            count += 1
+        return count
+
+    # ------------------------------------------------------------------
+    # Helpers
+    # ------------------------------------------------------------------
+
+    def _scope_dir(self, entry: FedEntry) -> Path:
+        label = "guild" if entry.layer == "guild" else f"{entry.layer}:{entry.entity}"
+        return self.base_dir / label
+
+    def _concept_path(self, entry: FedEntry) -> Path:
+        folder = _KIND_FOLDER.get(entry.kind, "misc")
+        slug = (
+            re.sub(r"[/\\]", "-", entry.term.strip())
+            .lower()
+            .replace(" ", "_")
+            .replace(":", "-")
+        )
+        path = self._scope_dir(entry) / folder / f"{slug}.md"
+        if not path.resolve().is_relative_to(self.base_dir.resolve()):
+            raise ValueError(f"unsafe path derived from term: {entry.term!r}")
+        return path
+
+
+# ------------------------------------------------------------------
+# Serialization helpers (module-level for testability)
+# ------------------------------------------------------------------
+
+
+def _entry_to_md(entry: FedEntry) -> str:
+    """FedEntry → OKF .md 문자열 (SPEC §4.1)."""
+    fm: dict = {
+        "type": entry.kind.capitalize() if entry.kind else "Concept",
+        "title": entry.term,
+        "description": entry.definition,
+    }
+    if entry.tags:
+        fm["tags"] = entry.tags
+    if entry.applies_to:
+        fm["applies_to"] = entry.applies_to
+    if entry.synonyms:
+        fm["synonyms"] = entry.synonyms
+    fm["layer"] = entry.layer
+    fm["entity"] = entry.entity
+    fm["inferred"] = entry.inferred
+    fm["timestamp"] = datetime.now(timezone.utc).isoformat()
+
+    yaml_block = yaml.dump(
+        fm, allow_unicode=True, default_flow_style=False, sort_keys=False
+    )
+    return f"---\n{yaml_block}---\n\n{entry.definition}\n"
+
+
+def _md_to_entry(path: Path) -> FedEntry | None:
+    """OKF .md 파일 → FedEntry. 파싱 실패 시 None 반환."""
+    text = path.read_text(encoding="utf-8")
+    if not text.startswith("---\n"):
+        return None
+    rest = text[4:]  # skip opening "---\n"
+    parts = rest.split("\n---\n", 1)
+    if len(parts) < 2:
+        return None
+    try:
+        fm = yaml.safe_load(parts[0])
+    except yaml.YAMLError:
+        return None
+    if not isinstance(fm, dict):
+        return None
+
+    raw_kind = str(fm.get("type", "")).lower()
+    kind = raw_kind if raw_kind in _KIND_FOLDER else ""
+
+    return FedEntry(
+        term=str(fm.get("title", path.stem)),
+        layer=str(fm.get("layer", "guild")),
+        entity=str(fm.get("entity", "")),
+        definition=str(fm.get("description", "")),
+        synonyms=fm.get("synonyms") or [],
+        inferred=bool(fm.get("inferred", False)),
+        kind=kind,
+        applies_to=str(fm.get("applies_to", "")),
+        tags=fm.get("tags") or [],
+    )
diff --git a/src/lang2sql/frontends/discord/bot.py b/src/lang2sql/frontends/discord/bot.py
index 29915f2..d16cc18 100644
--- a/src/lang2sql/frontends/discord/bot.py
+++ b/src/lang2sql/frontends/discord/bot.py
@@ -150,12 +150,41 @@ async def connect(interaction: discord.Interaction, dsn: str) -> None:
                 handlers.connect(to_identity(_interaction_context(interaction)), dsn),
             )
 
-        @tree.command(name="ingest", description="Propose definitions from a document")
-        async def ingest(interaction: discord.Interaction, ref: str) -> None:
+        @tree.command(
+            name="ingest",
+            description="문서에서 비즈니스 용어 후보 추출 (ref: 파일명, content: 텍스트 직접 입력)",
+        )
+        async def ingest(
+            interaction: discord.Interaction,
+            ref: str = "",
+            content: str = "",
+        ) -> None:
             await self._run(
                 interaction,
                 handlers.ingest(
-                    to_identity(_interaction_context(interaction)), ref=ref
+                    to_identity(_interaction_context(interaction)),
+                    ref=ref or None,
+                    content=content or None,
+                ),
+            )
+
+        @tree.command(
+            name="confirm_ingest",
+            description="ingest로 추출한 후보를 시멘틱 레이어에 등록",
+        )
+        async def confirm_ingest(
+            interaction: discord.Interaction,
+            ref: str,
+            accept: str = "all",
+            layer: str = "channel",
+        ) -> None:
+            await self._run(
+                interaction,
+                handlers.confirm_ingest(
+                    to_identity(_interaction_context(interaction)),
+                    ref=ref,
+                    accept=accept,
+                    layer=layer,
                 ),
             )
 
diff --git a/src/lang2sql/frontends/discord/commands.py b/src/lang2sql/frontends/discord/commands.py
index 4132706..7319fdc 100644
--- a/src/lang2sql/frontends/discord/commands.py
+++ b/src/lang2sql/frontends/discord/commands.py
@@ -247,6 +247,23 @@ async def ingest(
         result = await ctx.tools.dispatch("ingest_doc", args, ctx, "cmd:ingest")
         return OutboundMessage(text=result.content)
 
+    async def confirm_ingest(
+        self,
+        identity: Identity,
+        ref: str,
+        accept: str = "all",
+        layer: str = "channel",
+    ) -> OutboundMessage:
+        """ingest_doc로 추출한 후보를 검토 후 시멘틱 레이어에 등록."""
+        ctx = await self._concierge.build_context(identity)
+        result = await ctx.tools.dispatch(
+            "confirm_ingest",
+            {"ref": ref, "accept": accept, "layer": layer},
+            ctx,
+            "cmd:confirm_ingest",
+        )
+        return OutboundMessage(text=result.content)
+
 
 def _fmt_ts(ts: float) -> str:
     """Format an epoch timestamp as a short UTC string for audit listings."""
diff --git a/src/lang2sql/harness/context.py b/src/lang2sql/harness/context.py
index d9af9b1..a9fdb8c 100644
--- a/src/lang2sql/harness/context.py
+++ b/src/lang2sql/harness/context.py
@@ -33,4 +33,5 @@ class HarnessContext:
     safety: SafetyPipelinePort | None = None
     audit: AuditPort | None = None
     store: SqliteStore | None = None
+    okf_bundle_dir: str | None = None
     max_turns: int = 8
diff --git a/src/lang2sql/tenancy/concierge.py b/src/lang2sql/tenancy/concierge.py
index 628bd32..700116f 100644
--- a/src/lang2sql/tenancy/concierge.py
+++ b/src/lang2sql/tenancy/concierge.py
@@ -138,6 +138,7 @@ async def build_context(
             safety=self._safety,
             audit=self._audit,
             store=self._store,
+            okf_bundle_dir=os.getenv("OKF_BUNDLE_DIR"),
             max_turns=self._max_turns,
         )
 
diff --git a/src/lang2sql/tools/__init__.py b/src/lang2sql/tools/__init__.py
index 726b2c4..afc6dd6 100644
--- a/src/lang2sql/tools/__init__.py
+++ b/src/lang2sql/tools/__init__.py
@@ -13,6 +13,7 @@
 from ..ingestion.pipeline import IngestionPipeline
 from ..memory.service import MemoryService
 from .ask_user import AskUser
+from .confirm_ingest import ConfirmIngest
 from .enrich_schema import EnrichSchema
 from .explore_schema import ExploreSchema
 from .ingest_doc import IngestDoc
@@ -31,6 +32,7 @@
     "Remember",
     "AskUser",
     "IngestDoc",
+    "ConfirmIngest",
 ]
 
 
@@ -51,4 +53,5 @@ def build_default_tools(
         AskUser(),
         Remember(memory),
         IngestDoc(ingestion, source, extractor),
+        ConfirmIngest(),
     ]
diff --git a/src/lang2sql/tools/confirm_ingest.py b/src/lang2sql/tools/confirm_ingest.py
new file mode 100644
index 0000000..05bc5f1
--- /dev/null
+++ b/src/lang2sql/tools/confirm_ingest.py
@@ -0,0 +1,165 @@
+"""confirm_ingest — 사용자가 승인한 ingest 후보를 시멘틱 레이어에 등록.
+
+ingest_doc이 KV에 저장한 pending_ingest:{ref} 후보 목록을 읽어
+FedEntry로 변환하고 KV에 저장한다. OKF_BUNDLE_DIR 환경변수가 설정된 경우
+OkfBundle로도 내보낸다.
+"""
+
+from __future__ import annotations
+
+import json
+from typing import TYPE_CHECKING, Any
+
+from ..core.ports.ingestion import CandidateKind, SemanticCandidate
+from ..core.types import ToolResult, ToolSpec
+from ..tools.ingest_doc import PENDING_PREFIX
+from ..tools.semantic_federation import FedEntry, _kv_key, _validate_layer
+
+if TYPE_CHECKING:
+    from ..harness.context import HarnessContext
+
+
+def _dict_to_candidate(d: dict) -> SemanticCandidate:
+    return SemanticCandidate(
+        kind=CandidateKind(d["kind"]),
+        name=d["name"],
+        definition=d["definition"],
+        applies_to=d.get("applies_to", ""),
+        source_id=d.get("source_id", ""),
+    )
+
+
+class ConfirmIngest:
+    """pending_ingest 후보를 KV(+OkfBundle)에 등록하는 툴."""
+
+    @property
+    def spec(self) -> ToolSpec:
+        return ToolSpec(
+            name="confirm_ingest",
+            description=(
+                "Register approved semantic candidates from a previously ingested document "
+                "into the semantic layer (KV store). "
+                "Call ingest_doc first to extract candidates."
+            ),
+            parameters={
+                "type": "object",
+                "properties": {
+                    "ref": {
+                        "type": "string",
+                        "description": "document ref used with ingest_doc",
+                    },
+                    "accept": {
+                        "type": "string",
+                        "description": (
+                            "'all' to register every candidate, "
+                            "or comma-separated 1-based indices like '1,3'"
+                        ),
+                        "default": "all",
+                    },
+                    "layer": {
+                        "type": "string",
+                        "enum": ["guild", "channel", "member"],
+                        "description": "scope to register under (default: channel)",
+                        "default": "channel",
+                    },
+                },
+                "required": ["ref"],
+            },
+        )
+
+    async def run(self, args: dict[str, Any], ctx: "HarnessContext") -> ToolResult:
+        ref = (args.get("ref") or "").strip()
+        accept = (args.get("accept") or "all").strip()
+        layer_raw = (args.get("layer") or "channel").strip()
+
+        if not ref:
+            return ToolResult(call_id="", content="'ref' is required.", is_error=True)
+        if ctx.store is None:
+            return ToolResult(call_id="", content="No store available.", is_error=True)
+
+        channel_id = ctx.identity.effective_channel_id
+        layer, err = _validate_layer(layer_raw, channel_id, ctx.identity.is_admin)
+        if err:
+            return ToolResult(call_id="", content=err, is_error=True)
+
+        kv_scope = ctx.identity.kv_scope
+        pending_key = f"{PENDING_PREFIX}:{ref}"
+        raw = ctx.store.kv_get(kv_scope, pending_key)
+        if not raw:
+            return ToolResult(
+                call_id="",
+                content=f"No pending candidates for '{ref}'. Run ingest_doc first.",
+                is_error=True,
+            )
+
+        try:
+            all_candidates = [_dict_to_candidate(d) for d in json.loads(raw)]
+        except (json.JSONDecodeError, KeyError, ValueError) as exc:
+            return ToolResult(
+                call_id="",
+                content=f"Pending data corrupted: {exc}",
+                is_error=True,
+            )
+
+        selected = _select(all_candidates, accept)
+        if selected is None:
+            return ToolResult(
+                call_id="",
+                content="Invalid accept value. Use 'all' or comma-separated indices like '1,3'.",
+                is_error=True,
+            )
+        if not selected:
+            return ToolResult(call_id="", content="No candidates selected.")
+
+        entity = (
+            ""
+            if layer == "guild"
+            else (channel_id if layer == "channel" else ctx.identity.user_id)
+        )
+        registered: list[str] = []
+        for cand in selected:
+            entry = FedEntry(
+                term=cand.name,
+                layer=layer,
+                entity=entity,
+                definition=cand.definition,
+                inferred=False,
+                kind=cand.kind.value,
+                applies_to=cand.applies_to,
+            )
+            ctx.store.kv_set(
+                kv_scope,
+                _kv_key(entry.term, entry.layer, entry.entity),
+                entry.to_json(),
+            )
+            registered.append(entry.term)
+
+        ctx.store.kv_delete(kv_scope, pending_key)
+
+        if ctx.okf_bundle_dir and registered:
+            from ..adapters.storage.okf_bundle import OkfBundle
+
+            OkfBundle(ctx.okf_bundle_dir).export(ctx.store, kv_scope)
+
+        kind_labels = {c.name: c.kind.value.upper() for c in selected}
+        lines = [f"✅ {len(registered)} term(s) registered to [{layer}]:"]
+        for t in registered:
+            lines.append(f"  - [{kind_labels[t]}] {t}")
+        return ToolResult(call_id="", content="\n".join(lines))
+
+
+def _select(
+    candidates: list[SemanticCandidate], accept: str
+) -> list[SemanticCandidate] | None:
+    if accept == "all":
+        return list(candidates)
+    try:
+        indices = [int(i.strip()) - 1 for i in accept.split(",") if i.strip()]
+    except ValueError:
+        return None
+    result = []
+    for idx in indices:
+        if idx < 0 or idx >= len(candidates):
+            return None
+        result.append(candidates[idx])
+    return result
diff --git a/src/lang2sql/tools/ingest_doc.py b/src/lang2sql/tools/ingest_doc.py
index adc5ebf..85c573c 100644
--- a/src/lang2sql/tools/ingest_doc.py
+++ b/src/lang2sql/tools/ingest_doc.py
@@ -1,22 +1,35 @@
 """ingest_doc — turn an uploaded document into semantic candidates (★③).
 
 Runs the Source × Extractor pipeline and returns the proposed metric/rule
-definitions for the user to confirm. V1 does NOT auto-register — confirmation
-is a frontend step (Discord buttons in Week 4); this tool surfaces the
-candidates so the human stays in the loop (documents are the source of truth).
+definitions for the user to confirm. Candidates are stored in KV under a
+``pending_ingest:{ref}`` key so ``confirm_ingest`` can retrieve and register
+them once the user approves.
 """
 
 from __future__ import annotations
 
+import json
 from typing import TYPE_CHECKING, Any
 
-from ..core.ports.ingestion import DocExtractorPort, SourcePort
+from ..core.ports.ingestion import DocExtractorPort, SemanticCandidate, SourcePort
 from ..core.types import ToolResult, ToolSpec
 from ..ingestion.pipeline import IngestionPipeline
 
 if TYPE_CHECKING:
     from ..harness.context import HarnessContext
 
+PENDING_PREFIX = "pending_ingest"
+
+
+def _candidate_to_dict(c: SemanticCandidate) -> dict:
+    return {
+        "kind": c.kind.value,
+        "name": c.name,
+        "definition": c.definition,
+        "applies_to": c.applies_to,
+        "source_id": c.source_id,
+    }
+
 
 class IngestDoc:
     def __init__(
@@ -35,7 +48,8 @@ def spec(self) -> ToolSpec:
             name="ingest_doc",
             description=(
                 "Read a document and propose metric/dimension/rule definitions "
-                "for the user to confirm before they enter the semantic layer."
+                "for the user to confirm before they enter the semantic layer. "
+                "Use confirm_ingest to register the approved candidates."
             ),
             parameters={
                 "type": "object",
@@ -53,15 +67,19 @@ def spec(self) -> ToolSpec:
         )
 
     async def run(self, args: dict[str, Any], ctx: "HarnessContext") -> ToolResult:
-        ref = (args.get("ref") or "inline").strip()
+        ref = (args.get("ref") or "").strip()
         content = args.get("content")
         blob = content.encode("utf-8") if isinstance(content, str) else None
-        if not content and ref == "inline":
+        if not content and not ref:
             return ToolResult(
                 call_id="",
                 content="provide a document 'ref' or inline 'content'",
                 is_error=True,
             )
+        if not ref:
+            import hashlib
+
+            ref = "inline:" + hashlib.md5(blob or b"").hexdigest()[:8]
 
         candidates = await self._pipeline.ingest(
             self._source, self._extractor, ref, blob
@@ -71,8 +89,22 @@ async def run(self, args: dict[str, Any], ctx: "HarnessContext") -> ToolResult:
                 call_id="", content="No definitions found in the document."
             )
 
-        lines = ["Proposed definitions (confirm to register):"]
-        for c in candidates:
+        if ctx.store is not None:
+            pending_key = f"{PENDING_PREFIX}:{ref}"
+            ctx.store.kv_set(
+                ctx.identity.kv_scope,
+                pending_key,
+                json.dumps([_candidate_to_dict(c) for c in candidates]),
+            )
+
+        lines = [f"Proposed definitions from '{ref}' (use confirm_ingest to register):"]
+        for i, c in enumerate(candidates, 1):
             applies = f" [{c.applies_to}]" if c.applies_to else ""
-            lines.append(f"- {c.kind.value.upper()} {c.name}{applies} → {c.definition}")
+            lines.append(
+                f"  {i}. [{c.kind.value.upper()}] {c.name}{applies} — {c.definition}"
+            )
+        lines.append(
+            f"\nRun confirm_ingest(ref='{ref}', accept='all') to register all, "
+            "or specify indices like accept='1,3'."
+        )
         return ToolResult(call_id="", content="\n".join(lines))
diff --git a/src/lang2sql/tools/semantic_federation.py b/src/lang2sql/tools/semantic_federation.py
index f58a297..cdb2f30 100644
--- a/src/lang2sql/tools/semantic_federation.py
+++ b/src/lang2sql/tools/semantic_federation.py
@@ -28,6 +28,21 @@
 _KV_PREFIX = "cterm"
 _LAYERS = ("guild", "channel", "member")
 
+
+def _validate_layer(
+    layer_raw: str, channel_id: str, is_admin: bool
+) -> tuple[str, str | None]:
+    """layer 정규화·권한 검증. (normalized_layer, error_msg_or_None) 반환."""
+    layer = layer_raw.strip().lower()
+    if layer not in _LAYERS:
+        return layer, f"❌ layer는 {list(_LAYERS)} 중 하나여야 합니다."
+    if layer == "guild" and not is_admin:
+        return layer, "❌ guild 용어 등록·수정은 관리자만 가능합니다."
+    if layer == "channel" and not channel_id:
+        return layer, "❌ 채널 컨텍스트 없이 channel 레이어에 등록할 수 없습니다."
+    return layer, None
+
+
 from ..tools.enrich_schema import (
     _KV_PREFIX as _ENRICH_PREFIX,
     _KV_RELATIONSHIPS as _ENRICH_RELATIONSHIPS,
@@ -236,27 +251,11 @@ async def run(self, args: dict[str, Any], ctx: "HarnessContext") -> ToolResult:
                 call_id="", content=f"🗑️ **{term}** [{', '.join(deleted_tags)}] 삭제"
             )
 
-        layer = str(args.get("layer", "member")).strip().lower()
-        if layer not in _LAYERS:
-            return ToolResult(
-                call_id="",
-                content=f"❌ layer는 {list(_LAYERS)} 중 하나여야 합니다.",
-                is_error=True,
-            )
-
-        if layer == "guild" and not ctx.identity.is_admin:
-            return ToolResult(
-                call_id="",
-                content="❌ guild 용어 등록·수정은 관리자만 가능합니다.",
-                is_error=True,
-            )
-
-        if layer == "channel" and not channel_id:
-            return ToolResult(
-                call_id="",
-                content="❌ 채널 컨텍스트 없이 channel 레이어에 등록할 수 없습니다.",
-                is_error=True,
-            )
+        layer, err = _validate_layer(
+            str(args.get("layer", "member")), channel_id, ctx.identity.is_admin
+        )
+        if err:
+            return ToolResult(call_id="", content=err, is_error=True)
 
         entity = (
             "" if layer == "guild" else (user_id if layer == "member" else channel_id)
diff --git a/tests/test_confirm_ingest.py b/tests/test_confirm_ingest.py
new file mode 100644
index 0000000..4eb8671
--- /dev/null
+++ b/tests/test_confirm_ingest.py
@@ -0,0 +1,320 @@
+"""confirm_ingest — pending 후보 등록 및 OkfBundle 연동 테스트."""
+
+from __future__ import annotations
+
+import asyncio
+import json
+import tempfile
+from pathlib import Path
+from typing import Sequence
+
+from lang2sql.adapters.storage.sqlite_store import SqliteStore
+from lang2sql.core.identity import Identity
+from lang2sql.core.ports.ingestion import CandidateKind, SemanticCandidate
+from lang2sql.core.types import Completion, Message, ToolSpec
+from lang2sql.harness.context import HarnessContext
+from lang2sql.harness.session import Session
+from lang2sql.harness.tool_registry import ToolRegistry
+from lang2sql.tools import build_default_tools
+from lang2sql.tools.confirm_ingest import ConfirmIngest, _dict_to_candidate, _select
+from lang2sql.tools.ingest_doc import PENDING_PREFIX, IngestDoc, _candidate_to_dict
+from lang2sql.tools.semantic_federation import FedEntry, _kv_key
+
+# ---------------------------------------------------------------------------
+# Helpers
+# ---------------------------------------------------------------------------
+
+
+class _FakeLLM:
+    def __init__(self, content: str = "[]") -> None:
+        self._content = content
+
+    async def complete(
+        self, messages: Sequence[Message], tools: Sequence[ToolSpec] = ()
+    ) -> Completion:
+        return Completion(content=self._content, finish_reason="stop")
+
+
+def _make_ctx(
+    store: SqliteStore,
+    okf_bundle_dir: str | None = None,
+    is_admin: bool = False,
+    channel_id: str = "c1",
+) -> HarnessContext:
+    identity = Identity(
+        user_id="u1", guild_id="g1", channel_id=channel_id, is_admin=is_admin
+    )
+    from lang2sql.ingestion import FileSource, IngestionPipeline, LLMExtractor
+    from lang2sql.memory import (
+        InjectAllRecall,
+        InMemoryStore,
+        ManualExtractor,
+        MemoryService,
+    )
+
+    memory = MemoryService(InMemoryStore(), InjectAllRecall(), ManualExtractor())
+    ingestion = IngestionPipeline()
+    source = FileSource()
+    extractor = LLMExtractor(_FakeLLM())
+    tools = ToolRegistry(
+        build_default_tools(
+            memory=memory, ingestion=ingestion, source=source, extractor=extractor
+        )
+    )
+    return HarnessContext(
+        identity=identity,
+        llm=_FakeLLM(),
+        tools=tools,
+        session=Session(identity=identity),
+        store=store,
+        okf_bundle_dir=okf_bundle_dir,
+    )
+
+
+def _seed_pending(
+    store: SqliteStore, scope: str, ref: str, candidates: list[SemanticCandidate]
+) -> None:
+    key = f"{PENDING_PREFIX}:{ref}"
+    store.kv_set(scope, key, json.dumps([_candidate_to_dict(c) for c in candidates]))
+
+
+_SAMPLE = [
+    SemanticCandidate(
+        CandidateKind.METRIC,
+        "monthly_revenue",
+        "SUM(orders.amount)",
+        applies_to="orders",
+    ),
+    SemanticCandidate(CandidateKind.RULE, "exclude_cancelled", "status != 'cancelled'"),
+    SemanticCandidate(CandidateKind.DIMENSION, "customer_tier", "users.tier"),
+]
+
+
+# ---------------------------------------------------------------------------
+# 직렬화 단위 테스트
+# ---------------------------------------------------------------------------
+
+
+def test_candidate_roundtrip() -> None:
+    for c in _SAMPLE:
+        d = _candidate_to_dict(c)
+        restored = _dict_to_candidate(d)
+        assert restored.kind == c.kind
+        assert restored.name == c.name
+        assert restored.definition == c.definition
+
+
+def test_select_all() -> None:
+    assert _select(_SAMPLE, "all") == _SAMPLE
+
+
+def test_select_indices() -> None:
+    result = _select(_SAMPLE, "1,3")
+    assert result is not None
+    assert [c.name for c in result] == ["monthly_revenue", "customer_tier"]
+
+
+def test_select_out_of_range_returns_none() -> None:
+    assert _select(_SAMPLE, "9") is None
+
+
+def test_select_invalid_string_returns_none() -> None:
+    assert _select(_SAMPLE, "foo") is None
+
+
+# ---------------------------------------------------------------------------
+# confirm_ingest 동작 테스트
+# ---------------------------------------------------------------------------
+
+
+def test_confirm_all_saves_fed_entries() -> None:
+    store = SqliteStore()
+    ctx = _make_ctx(store, is_admin=True)
+    scope = ctx.identity.kv_scope
+    _seed_pending(store, scope, "defs.md", _SAMPLE)
+
+    tool = ConfirmIngest()
+    result = asyncio.run(
+        tool.run({"ref": "defs.md", "accept": "all", "layer": "guild"}, ctx)
+    )
+
+    assert not result.is_error
+    assert "3 term(s)" in result.content
+
+    for cand in _SAMPLE:
+        raw = store.kv_get(scope, _kv_key(cand.name, "guild", ""))
+        assert raw is not None
+        entry = FedEntry.from_json(raw)
+        assert entry.term == cand.name
+        assert entry.kind == cand.kind.value
+        assert entry.layer == "guild"
+
+
+def test_confirm_by_index_saves_selected_only() -> None:
+    store = SqliteStore()
+    ctx = _make_ctx(store, is_admin=True)
+    scope = ctx.identity.kv_scope
+    _seed_pending(store, scope, "defs.md", _SAMPLE)
+
+    result = asyncio.run(
+        ConfirmIngest().run({"ref": "defs.md", "accept": "2", "layer": "guild"}, ctx)
+    )
+
+    assert not result.is_error
+    assert "1 term(s)" in result.content
+
+    assert store.kv_get(scope, _kv_key("exclude_cancelled", "guild", "")) is not None
+    assert store.kv_get(scope, _kv_key("monthly_revenue", "guild", "")) is None
+
+
+def test_confirm_channel_layer_uses_channel_entity() -> None:
+    store = SqliteStore()
+    ctx = _make_ctx(store)
+    scope = ctx.identity.kv_scope
+    _seed_pending(store, scope, "defs.md", [_SAMPLE[0]])
+
+    asyncio.run(
+        ConfirmIngest().run(
+            {"ref": "defs.md", "accept": "all", "layer": "channel"}, ctx
+        )
+    )
+
+    ch_id = ctx.identity.effective_channel_id
+    raw = store.kv_get(scope, _kv_key("monthly_revenue", "channel", ch_id))
+    assert raw is not None
+    entry = FedEntry.from_json(raw)
+    assert entry.entity == ch_id
+
+
+def test_confirm_member_layer_uses_user_id() -> None:
+    store = SqliteStore()
+    ctx = _make_ctx(store)
+    scope = ctx.identity.kv_scope
+    _seed_pending(store, scope, "defs.md", [_SAMPLE[0]])
+
+    asyncio.run(
+        ConfirmIngest().run({"ref": "defs.md", "accept": "all", "layer": "member"}, ctx)
+    )
+
+    raw = store.kv_get(
+        scope, _kv_key("monthly_revenue", "member", ctx.identity.user_id)
+    )
+    assert raw is not None
+
+
+def test_confirm_clears_pending_key_after_success() -> None:
+    store = SqliteStore()
+    ctx = _make_ctx(store, is_admin=True)
+    scope = ctx.identity.kv_scope
+    _seed_pending(store, scope, "defs.md", [_SAMPLE[0]])
+
+    asyncio.run(
+        ConfirmIngest().run({"ref": "defs.md", "accept": "all", "layer": "guild"}, ctx)
+    )
+
+    assert store.kv_get(scope, f"{PENDING_PREFIX}:defs.md") is None
+
+
+def test_confirm_guild_layer_blocked_for_non_admin() -> None:
+    store = SqliteStore()
+    ctx = _make_ctx(store, is_admin=False)
+    scope = ctx.identity.kv_scope
+    _seed_pending(store, scope, "defs.md", [_SAMPLE[0]])
+
+    result = asyncio.run(
+        ConfirmIngest().run({"ref": "defs.md", "accept": "all", "layer": "guild"}, ctx)
+    )
+    assert result.is_error
+    assert "관리자" in result.content
+
+
+def test_confirm_channel_layer_blocked_without_channel_id() -> None:
+    store = SqliteStore()
+    ctx = _make_ctx(store, channel_id="")
+    scope = ctx.identity.kv_scope
+    _seed_pending(store, scope, "defs.md", [_SAMPLE[0]])
+
+    result = asyncio.run(
+        ConfirmIngest().run(
+            {"ref": "defs.md", "accept": "all", "layer": "channel"}, ctx
+        )
+    )
+    assert result.is_error
+    assert "channel" in result.content
+
+
+def test_confirm_invalid_layer_returns_error() -> None:
+    store = SqliteStore()
+    ctx = _make_ctx(store)
+    scope = ctx.identity.kv_scope
+    _seed_pending(store, scope, "defs.md", [_SAMPLE[0]])
+
+    result = asyncio.run(
+        ConfirmIngest().run({"ref": "defs.md", "accept": "all", "layer": "team"}, ctx)
+    )
+    assert result.is_error
+    assert "layer" in result.content
+
+
+def test_confirm_missing_ref_returns_error() -> None:
+    store = SqliteStore()
+    ctx = _make_ctx(store)
+    result = asyncio.run(ConfirmIngest().run({"ref": "nonexistent.md"}, ctx))
+    assert result.is_error
+    assert "ingest_doc" in result.content
+
+
+def test_confirm_no_ref_arg_returns_error() -> None:
+    store = SqliteStore()
+    ctx = _make_ctx(store)
+    result = asyncio.run(ConfirmIngest().run({}, ctx))
+    assert result.is_error
+
+
+# ---------------------------------------------------------------------------
+# ingest_doc → confirm_ingest 전체 연동 테스트
+# ---------------------------------------------------------------------------
+
+
+def test_ingest_doc_saves_pending_key() -> None:
+    store = SqliteStore()
+    ctx = _make_ctx(store)
+    scope = ctx.identity.kv_scope
+
+    candidates = [SemanticCandidate(CandidateKind.METRIC, "active_user", "30d login")]
+    _seed_pending(store, scope, "test.md", candidates)
+
+    raw = store.kv_get(scope, f"{PENDING_PREFIX}:test.md")
+    assert raw is not None
+    loaded = json.loads(raw)
+    assert loaded[0]["name"] == "active_user"
+
+
+def test_confirm_with_okf_bundle_exports_files() -> None:
+    store = SqliteStore()
+    with tempfile.TemporaryDirectory() as bundle_dir:
+        ctx = _make_ctx(store, okf_bundle_dir=bundle_dir, is_admin=True)
+        scope = ctx.identity.kv_scope
+        _seed_pending(store, scope, "defs.md", [_SAMPLE[0]])
+
+        result = asyncio.run(
+            ConfirmIngest().run(
+                {"ref": "defs.md", "accept": "all", "layer": "guild"}, ctx
+            )
+        )
+        assert not result.is_error
+
+        md_files = list(Path(bundle_dir).rglob("*.md"))
+        assert len(md_files) >= 1
+        assert any("monthly_revenue" in f.name for f in md_files)
+
+
+def test_registered_tool_name_in_registry() -> None:
+    from lang2sql.tenancy.concierge import ContextConcierge
+
+    concierge = ContextConcierge()
+    identity = Identity(user_id="u1", guild_id="g1", channel_id="c1")
+    ctx = asyncio.run(concierge.build_context(identity))
+    names = {s.name for s in ctx.tools.specs()}
+    assert "confirm_ingest" in names
+    assert "ingest_doc" in names
diff --git a/tests/test_integration.py b/tests/test_integration.py
index d1295a6..79bb110 100644
--- a/tests/test_integration.py
+++ b/tests/test_integration.py
@@ -35,6 +35,7 @@ def test_v1_tools_registered():
         "ask_user",
         "remember",
         "ingest_doc",
+        "confirm_ingest",
     }
 
 
diff --git a/tests/test_okf_bundle.py b/tests/test_okf_bundle.py
new file mode 100644
index 0000000..99c2ec7
--- /dev/null
+++ b/tests/test_okf_bundle.py
@@ -0,0 +1,220 @@
+"""OkfBundle — export/import round-trip 및 파일 구조 테스트."""
+
+from __future__ import annotations
+
+import tempfile
+from pathlib import Path
+
+from lang2sql.adapters.storage.okf_bundle import OkfBundle, _entry_to_md, _md_to_entry
+from lang2sql.adapters.storage.sqlite_store import SqliteStore
+from lang2sql.tools.semantic_federation import FedEntry, _kv_key
+
+
+def _populate(store: SqliteStore, scope: str, entries: list[FedEntry]) -> None:
+    for e in entries:
+        store.kv_set(scope, _kv_key(e.term, e.layer, e.entity), e.to_json())
+
+
+# ------------------------------------------------------------------
+# 직렬화 단위 테스트
+# ------------------------------------------------------------------
+
+
+def test_entry_to_md_contains_required_okf_fields() -> None:
+    entry = FedEntry(
+        term="활성고객",
+        layer="guild",
+        entity="",
+        definition="30일 내 로그인한 users",
+        kind="metric",
+        applies_to="users",
+        tags=["growth"],
+    )
+    md = _entry_to_md(entry)
+    assert "type: Metric" in md
+    assert "title: 활성고객" in md
+    assert "description:" in md
+    assert "layer: guild" in md
+
+
+def test_md_to_entry_roundtrip() -> None:
+    entry = FedEntry(
+        term="순매출",
+        layer="channel",
+        entity="mkt-123",
+        definition="환불 제외 매출",
+        synonyms=["net revenue"],
+        kind="metric",
+        applies_to="orders",
+        tags=["finance"],
+        inferred=True,
+    )
+    with tempfile.NamedTemporaryFile(
+        suffix=".md", mode="w", delete=False, encoding="utf-8"
+    ) as f:
+        f.write(_entry_to_md(entry))
+        tmp = Path(f.name)
+
+    restored = _md_to_entry(tmp)
+    assert restored is not None
+    assert restored.term == "순매출"
+    assert restored.kind == "metric"
+    assert restored.layer == "channel"
+    assert restored.entity == "mkt-123"
+    assert restored.applies_to == "orders"
+    assert restored.tags == ["finance"]
+    assert restored.inferred is True
+    tmp.unlink()
+
+
+def test_md_to_entry_unknown_type_becomes_empty_kind() -> None:
+    md = "---\ntype: Playbook\ntitle: foo\ndescription: bar\nlayer: guild\nentity: ''\ninferred: false\n---\n\nbar\n"
+    with tempfile.NamedTemporaryFile(
+        suffix=".md", mode="w", delete=False, encoding="utf-8"
+    ) as f:
+        f.write(md)
+        tmp = Path(f.name)
+    entry = _md_to_entry(tmp)
+    assert entry is not None
+    assert entry.kind == ""
+    tmp.unlink()
+
+
+def test_md_to_entry_no_frontmatter_returns_none() -> None:
+    with tempfile.NamedTemporaryFile(
+        suffix=".md", mode="w", delete=False, encoding="utf-8"
+    ) as f:
+        f.write("no frontmatter here")
+        tmp = Path(f.name)
+    assert _md_to_entry(tmp) is None
+    tmp.unlink()
+
+
+# ------------------------------------------------------------------
+# export / import 통합 테스트
+# ------------------------------------------------------------------
+
+
+def test_export_creates_kind_based_folders() -> None:
+    store = SqliteStore()
+    scope = "g1"
+    entries = [
+        FedEntry("활성고객", "guild", "", "30일 로그인", kind="metric"),
+        FedEntry("orders", "guild", "", "주문 테이블", kind="table"),
+        FedEntry("환불제외", "guild", "", "status != refunded", kind="rule"),
+        FedEntry("고객등급", "guild", "", "users.tier", kind="dimension"),
+        FedEntry("기타용어", "guild", "", "정의 없음", kind=""),
+    ]
+    _populate(store, scope, entries)
+
+    with tempfile.TemporaryDirectory() as tmp:
+        bundle = OkfBundle(tmp)
+        count = bundle.export(store, scope)
+
+        assert count == 5
+        assert (Path(tmp) / "guild" / "metrics" / "활성고객.md").exists()
+        assert (Path(tmp) / "guild" / "tables" / "orders.md").exists()
+        assert (Path(tmp) / "guild" / "rules" / "환불제외.md").exists()
+        assert (Path(tmp) / "guild" / "dimensions" / "고객등급.md").exists()
+        assert (Path(tmp) / "guild" / "misc" / "기타용어.md").exists()
+
+
+def test_export_separates_scopes() -> None:
+    store = SqliteStore()
+    scope = "g1"
+    _populate(
+        store,
+        scope,
+        [
+            FedEntry("활성고객", "guild", "", "30일 로그인", kind="metric"),
+            FedEntry("활성고객", "channel", "mkt", "7일 구매", kind="metric"),
+        ],
+    )
+
+    with tempfile.TemporaryDirectory() as tmp:
+        bundle = OkfBundle(tmp)
+        bundle.export(store, scope)
+
+        assert (Path(tmp) / "guild" / "metrics" / "활성고객.md").exists()
+        assert (Path(tmp) / "channel:mkt" / "metrics" / "활성고객.md").exists()
+
+
+def test_import_restores_kv_from_files() -> None:
+    store = SqliteStore()
+    scope = "g1"
+    original = FedEntry(
+        "순매출",
+        "guild",
+        "",
+        "환불 제외 매출",
+        kind="metric",
+        applies_to="orders",
+        tags=["finance"],
+    )
+    _populate(store, scope, [original])
+
+    with tempfile.TemporaryDirectory() as tmp:
+        bundle = OkfBundle(tmp)
+        bundle.export(store, scope)
+
+        # KV 비우고 import
+        empty_store = SqliteStore()
+        count = bundle.import_(empty_store, scope)
+
+        assert count == 1
+        key = _kv_key("순매출", "guild", "")
+        raw = empty_store.kv_get(scope, key)
+        assert raw is not None
+        restored = FedEntry.from_json(raw)
+        assert restored.term == "순매출"
+        assert restored.kind == "metric"
+        assert restored.applies_to == "orders"
+
+
+def test_import_skips_reserved_files() -> None:
+    with tempfile.TemporaryDirectory() as tmp:
+        guild_dir = Path(tmp) / "guild"
+        guild_dir.mkdir()
+        (guild_dir / "index.md").write_text("# index", encoding="utf-8")
+        (guild_dir / "log.md").write_text("# log", encoding="utf-8")
+
+        store = SqliteStore()
+        bundle = OkfBundle(tmp)
+        count = bundle.import_(store, "g1")
+        assert count == 0
+
+
+def test_full_roundtrip_preserves_all_fields() -> None:
+    store = SqliteStore()
+    scope = "g1"
+    original = FedEntry(
+        term="월매출",
+        layer="member",
+        entity="user-99",
+        definition="당월 발생 매출 합계",
+        synonyms=["monthly revenue"],
+        inferred=False,
+        kind="metric",
+        applies_to="orders.amount",
+        tags=["finance", "monthly"],
+    )
+    _populate(store, scope, [original])
+
+    with tempfile.TemporaryDirectory() as tmp:
+        bundle = OkfBundle(tmp)
+        bundle.export(store, scope)
+        restored_store = SqliteStore()
+        bundle.import_(restored_store, scope)
+
+    key = _kv_key("월매출", "member", "user-99")
+    raw = restored_store.kv_get(scope, key)
+    assert raw is not None
+    restored = FedEntry.from_json(raw)
+
+    assert restored.term == "월매출"
+    assert restored.layer == "member"
+    assert restored.entity == "user-99"
+    assert restored.kind == "metric"
+    assert restored.applies_to == "orders.amount"
+    assert set(restored.tags) == {"finance", "monthly"}
+    assert restored.synonyms == ["monthly revenue"]