# Phase 13 — 외부 연동 & 데이터 커넥터 (mock → real)

> 한 줄 요약: phase 7~12에서 **mock-first** 로 만든 모든 페이지(라이프·메일·일정·알림·리서치·여행·결재함·자동화)를, 코드 한 줄 안 바꾸고 **실제 외부 데이터**로 전환한다. 커넥터 추상화(`backend/app/connectors/`) 뒤에 `MockConnector` 대신 `RealConnector` 를 꽂으면(`CONNECTOR_<DOMAIN>=real`), 동일 인터페이스(`fetch / sync / normalize / write`)를 통해 실데이터가 흘러들고 연합 이벤트(`mail.received`, `calendar.updated`, …)가 **실제로** 발생해 자동화·작업·알림이 진짜 데이터로 작동한다.

> 이 문서는 **포스트-MVP 세트의 일부**입니다 — 먼저 `dev/overview.md` 와 `dev/post-mvp-overview.md` 를 읽으세요.
>
> 선행(필수): `phase-2-backend.md`(모델/스키마/라우터/시드/Provider 추상화 규약 — 모든 신규 테이블·엔드포인트는 이 패턴을 따른다), `phase-7-approvals-automation.md`(자율성 코어·event_bus·자동화 엔진·결재함), `phase-8-calendar-meetings.md`, `phase-9-mail-notifications.md`, `phase-10-research-travel.md`, `phase-11-life-care.md`, `phase-12-daily-narrative.md`(모두 mock-first 로 커넥터를 통해 데이터를 읽도록 구현되어 있어야 한다).
>
> 후속: `phase-14-proactive-agent.md`(능동형 에이전트 + 멀티모달 캡처 — 본 phase 의 RealConnector 가 공급하는 실데이터 이벤트 위에서 동작).

---

## 0. 목차

1. [개요 & 목표](#1-개요--목표)
2. [선행 조건 / 산출물](#2-선행-조건--산출물)
3. [상세 구현 — 커넥터 프레임워크](#3-상세-구현--커넥터-프레임워크)
4. [데이터/타입/API 계약](#4-데이터타입api-계약)
5. [디자인 충실도 노트](#5-디자인-충실도-노트)
6. [상태 처리 & 엣지 케이스](#6-상태-처리--엣지-케이스)
7. [연합 이벤트(발행/구독)](#7-연합-이벤트발행구독)
8. [테스팅 & 검증](#8-테스팅--검증)
9. [완료 기준 (Definition of Done)](#9-완료-기준-definition-of-done)
10. [다음 단계](#10-다음-단계)

---

## 1. 개요 & 목표

### 1.1 이 phase가 끝나면 무엇이 동작하는가

phase 7~12 의 페이지들은 **이미** 동작한다 — 단, 시드(`life-data.js`/`mail-data.js`/`cal-data.js` 등에서 이식한 목업) 위에서. Phase 13 이 끝나면:

- `backend/app/connectors/` 에 도메인별 **base 인터페이스**(`fetch / sync / normalize / write`) + 각 도메인의 `MockConnector`(시드 기반) + `RealConnector`(실제 제공자) 가 존재한다.
- 환경변수 한 줄(`CONNECTOR_MAIL=real`, `CONNECTOR_CALENDAR=real`, …)만 바꾸면 **페이지/라우터 코드는 그대로** 두고 데이터 소스가 mock→real 로 전환된다(인터페이스 불변).
- 6개 도메인 커넥터가 준비된다: **mail**(Gmail API / Google Workspace / HEY), **calendar**(Google Calendar API), **chat**(메신저), **finance**(카드/뱅크 아그리게이터 또는 수동 CSV 임포트), **health**(Apple Health/HealthKit export / Google Fit), **knowledge**(Notion API / Readwise / 웹 클리퍼).
- 제공자별 OAuth2 플로우(또는 토큰/CSV 임포트 대안)와 **암호화된 토큰 저장**(`connector_account.token_enc`)·만료/갱신이 구현된다.
- **증분 sync**(델타 토큰/`since` 커서) · **정규화**(provider 페이로드 → 내부 모델) · **충돌/중복 처리**(`external_id` UNIQUE upsert) · **레이트리밋**(지수 백오프 + per-connector 윈도우)가 동작한다.
- 실데이터가 들어오면 `event_bus`(phase-7 정본)에 `mail.received` / `calendar.updated` 등 **실제 이벤트가 발행**되어, 자동화 엔진(`evaluator`)·작업 실체화·알림 트리아지가 **실데이터로** 동작한다.
- **로컬 우선**: Ollama 로컬 LLM 철학과 정합 — 가능한 데이터는 로컬/수동 임포트(CSV·HealthKit export·`.ics` 파일·Markdown 노트)로도 들일 수 있고, 클라우드 OAuth 연동은 **선택**이다.
- 라이프 페이지의 **연결된 데이터 소스 카드**(`life-data.js: sources` 의 `on`/`last` — "Apple Health · 방금 동기화", "우리카드 · 오늘 09:12", "Google Fit · 연결 안 됨")가 mock 의 정적 값이 아니라 **실제 커넥터 상태**(connected/disconnected/syncing/error/token_expired)를 비춘다.

### 1.2 핵심 설계 원칙 — "인터페이스 불변, 구현만 교체"

```
                                                      ┌─ MockConnector  (시드 기반, 기본값)
페이지/라우터 ── ConnectorRegistry.get(domain) ──→ BaseConnector(추상) ─┤
   (코드 불변)        (CONNECTOR_<DOMAIN> 로 선택)                       └─ RealConnector  (OAuth/실제 API)
                                                                              │ normalize()
                                                                              ▼
                                                                       내부 모델(email/event/finance_tx/...)
                                                                              │ upsert(external_id UNIQUE)
                                                                              ▼
                                                                       event_bus.publish(mail.received 등)
                                                                              │
                                                                              ▼
                                                                       자동화/작업/알림 (실데이터로 동작)
```

CONTRACT 의 횡단 아키텍처 명명을 그대로 사용한다: 커넥터는 `backend/app/connectors/`, 도메인은 `mail/calendar/chat/finance/health/knowledge`, 설정은 `CONNECTOR_<DOMAIN>=mock|real`(기본 `mock`), 이벤트는 `event_bus`(phase-7), 임베딩/지식은 `backend/app/rag/`(phase-10) 와 공유한다.

> **모델 비종속 = 제공자 비종속의 확장.** phase-2 가 LLM 을 `LLMProvider` 로 추상화해 "모델 강제 금지 + env 주입 + 폴백" 한 것과 동일한 정신으로, Phase 13 은 외부 데이터 제공자를 `BaseConnector` 로 추상화해 "**제공자 강제 금지 + env 주입 + mock 폴백**" 한다. Gmail/Google Calendar/Apple Health/Notion 등 구체 제공자는 **목표 예시이며 교체 가능**(CONTRACT 가정).

---

## 2. 선행 조건 / 산출물

### 2.1 선행 조건 (의존 phase)

| 의존 | 무엇을 가져오는가 |
|---|---|
| `phase-2-backend.md` | SQLModel/Pydantic/Alembic 패턴, 라우터 prefix 규약(라우터 내부 prefix 없음 → `main.py` 의 `include_router(prefix="/api")`), `config.py` Settings(env 주입), TEXT PK 규약, `run_seed(session=None, reset=True)`, `get_provider()`(LLM 추상화 — 폴백 패턴의 원형). |
| `phase-7-approvals-automation.md` | **`event_bus`**(내부 이벤트 버스, 발행/구독 정본), `evaluator`(이벤트→규칙 매칭), `approval` 모델 + `autonomy_setting`(low+mixed 이상=자동실행, high=승인대기), `backend/app/worker/`(스케줄러 — 동기화 작업을 여기에 등록). |
| `phase-8-calendar-meetings.md` | `calendar`/`event`/`focus_block`/`meeting` 모델. RealConnector(calendar) 의 정규화 타깃. |
| `phase-9-mail-notifications.md` | `mail_account`/`email`/`notification` 모델 + 메일→작업/일정 추출(`email.ai.tasks/events`), 알림 트리아지. RealConnector(mail) 의 정규화 타깃. |
| `phase-10-research-travel.md` | `backend/app/rag/`(ingest→chunk→embed→vector), `knowledge_qa`. knowledge 커넥터가 ingest 로 흘려보낸다. |
| `phase-11-life-care.md` | `connector_source`(health/finance/knowledge on/last), `health`/`finance`/`knowledge_item` 모델. 라이프 페이지 connector 카드. |
| `phase-12-daily-narrative.md` | 하루 마감 집계 — sync 결과(아낀 시간·자동 처리 건수)가 narrative 에 합산. |

> **이 phase 가 mock 을 끄지 않는다.** mock 은 데모/오프라인/CI 의 결정적 기본값으로 **영구 유지**된다. real 은 사용자가 명시적으로 켤 때만(`CONNECTOR_<DOMAIN>=real` + 자격증명) 활성화. 단일 데모 사용자(지우)는 유지하고 멀티유저/per-user 토큰 스코프는 `phase-15-production.md` 에서 도입한다(본 phase 의 `connector_account.user_id` 는 `"jiwoo"` 고정 시드).

### 2.2 산출물 (Deliverables)

```
backend/
├─ app/
│  ├─ config.py                      (+) CONNECTOR_<DOMAIN>, OAuth/자격증명 env, ARI_SECRET_KEY
│  ├─ models.py                      (+) connector_account, connector_sync_state, external_link, oauth_state, connector_account_log
│  ├─ schemas.py                     (+) ConnectorStatusOut, OAuthStartOut, SyncResultOut, ImportResultOut, ...
│  ├─ crypto.py                      (+) Fernet 기반 토큰 암복호화 (ARI_SECRET_KEY 파생)
│  ├─ connectors/
│  │  ├─ __init__.py
│  │  ├─ base.py                     BaseConnector(추상): fetch/sync/normalize/write + 공통 upsert/레이트리밋
│  │  ├─ registry.py                 ConnectorRegistry.get(domain) — CONNECTOR_<DOMAIN> 로 mock|real 선택 + 캐시
│  │  ├─ ratelimit.py               토큰 버킷 + 지수 백오프(429/Retry-After)
│  │  ├─ oauth.py                    OAuth2 Authorization Code(+PKCE) 공통 헬퍼 (start/callback/refresh)
│  │  ├─ mail/        base.py mock.py real_gmail.py real_imap.py normalize.py
│  │  ├─ calendar/    base.py mock.py real_google.py ics_import.py normalize.py
│  │  ├─ chat/        base.py mock.py real_webhook.py normalize.py
│  │  ├─ finance/     base.py mock.py csv_import.py real_aggregator.py normalize.py
│  │  ├─ health/      base.py mock.py healthkit_import.py real_fit.py normalize.py
│  │  └─ knowledge/   base.py mock.py real_notion.py real_readwise.py clipper.py normalize.py
│  ├─ routers/
│  │  └─ connectors.py               GET/POST /api/connectors* (상태/연결/해제/동기화/임포트/OAuth 콜백)
│  └─ worker/
│     └─ sync_jobs.py                (+) 주기 sync 잡 등록(APScheduler) + 수동 트리거 진입점
├─ migrations/versions/xxxx_connectors.py
└─ tests/
   ├─ test_connector_contract.py     모든 커넥터가 BaseConnector 계약 충족(인터페이스 계약 테스트)
   ├─ test_connector_swap.py         CONNECTOR_<DOMAIN> 스왑(mock↔real) — 라우터 응답 동일 shape
   ├─ test_oauth_mock.py             OAuth 플로우 모킹(respx) — start→callback→token 저장(암호화)
   ├─ test_sync_idempotent.py        동기화 멱등성(같은 페이로드 2회 → 중복 0)
   ├─ test_normalize_golden.py       정규화 골든(provider 페이로드 → 내부 모델 고정 매핑)
   ├─ test_crypto.py                 토큰 암복호화 왕복
   └─ test_federation_real.py        real(가짜) sync → event_bus 에 mail.received/calendar.updated 발행
frontend/
├─ lib/types.ts                      (+) ConnectorStatus, OAuthStart, SyncResult, ImportResult
├─ components/connectors/
│  ├─ ConnectorCard.tsx              라이프 connector 카드 재사용 (on/last/상태)
│  ├─ ConnectStateBadge.tsx          연결/해제/동기화 중/오류/권한 만료 배지
│  └─ ImportDropzone.tsx             CSV/.ics/HealthKit export 수동 임포트
└─ playwright/connectors.spec.ts     연결→동기화→실데이터 반영 E2E (real=fake provider)
```

> 새 의존성(backend): `cryptography`(Fernet 토큰 암호화), `respx`(httpx OAuth 모킹 — dev). pip/uv 추가는 §8.1 에 명령 포함. **APScheduler 는 phase-7 worker 에서 이미 도입**했다고 가정(없으면 phase-7 에서 추가).

---

## 3. 상세 구현 — 커넥터 프레임워크

### 3.0 설정 — `config.py` 확장

phase-2 `Settings` 에 커넥터/OAuth/암호화 키 필드를 더한다. pydantic-settings 가 필드명을 대문자 env 로 자동 매핑(`connector_mail` → `CONNECTOR_MAIL`).

```python
# backend/app/config.py  (phase-2 Settings 에 추가)
from functools import lru_cache
from pydantic_settings import BaseSettings, SettingsConfigDict


class Settings(BaseSettings):
    model_config = SettingsConfigDict(env_file=".env", extra="ignore")

    # ── (phase-2 에서 상속) ──
    database_url: str = "sqlite:///./ari.db"
    ollama_host: str = "http://localhost:11434"
    ollama_model: str = "llama3.1"
    llm_provider: str = "auto"
    llm_timeout: float = 20.0
    risk_today: int = 8
    frontend_origin: str = "http://localhost:3000"
    ari_allow_test_reset: bool = False

    # ── (phase-13 신규) 도메인별 커넥터 선택. 기본 mock. ──
    connector_mail: str = "mock"          # mock | real
    connector_calendar: str = "mock"
    connector_chat: str = "mock"
    connector_finance: str = "mock"       # mock | csv | real
    connector_health: str = "mock"        # mock | healthkit | real
    connector_knowledge: str = "mock"

    # ── 토큰 암호화 키(필수 when real). Fernet 키 파생용. ──
    ari_secret_key: str = "dev-insecure-change-me-please-32bytes!"

    # ── OAuth2 자격증명(제공자별). 비어 있으면 해당 RealConnector 비활성. ──
    google_client_id: str = ""            # Gmail + Google Calendar 공용
    google_client_secret: str = ""
    google_redirect_uri: str = "http://localhost:8000/api/connectors/oauth/callback"
    notion_client_id: str = ""
    notion_client_secret: str = ""
    readwise_token: str = ""              # Readwise 는 단순 토큰(OAuth 아님)

    # ── 동기화 정책 ──
    sync_interval_minutes: int = 15       # worker 주기 sync
    sync_page_size: int = 50
    connector_http_timeout: float = 20.0


@lru_cache
def get_settings() -> Settings:
    return Settings()
```

`.env` 예시(루트 또는 `backend/.env`) — **로컬 우선이라 전부 mock 이 기본**, real 은 선택:

```bash
# (phase-0/2 에서 상속)
DATABASE_URL=sqlite:///./ari.db
OLLAMA_HOST=http://localhost:11434
OLLAMA_MODEL=qwen2.5:7b
LLM_PROVIDER=auto
FRONTEND_ORIGIN=http://localhost:3000

# ── 커넥터(phase-13) — 기본 전부 mock. 켤 도메인만 바꾼다. ──
CONNECTOR_MAIL=mock           # mock | real
CONNECTOR_CALENDAR=mock       # mock | real (real=Google Calendar)
CONNECTOR_CHAT=mock
CONNECTOR_FINANCE=mock        # mock | csv (수동 CSV 임포트) | real
CONNECTOR_HEALTH=mock         # mock | healthkit (export 임포트) | real (Google Fit)
CONNECTOR_KNOWLEDGE=mock      # mock | real (Notion/Readwise/클리퍼)

# ── 암호화 키(real 켤 때 필수, 32+ 바이트 권장) ──
ARI_SECRET_KEY=please-generate-a-long-random-secret-key

# ── OAuth (real 켤 때만) ──
GOOGLE_CLIENT_ID=
GOOGLE_CLIENT_SECRET=
GOOGLE_REDIRECT_URI=http://localhost:8000/api/connectors/oauth/callback
NOTION_CLIENT_ID=
NOTION_CLIENT_SECRET=
READWISE_TOKEN=
```

---

### 3.1 데이터 모델 — `models.py` 추가 (phase-2 패턴)

phase-2 규약 그대로: 모든 PK 는 TEXT(str), enum 은 `str, Enum`, `now()` 헬퍼 재사용. 신규 5개 테이블.

```python
# backend/app/models.py  (phase-2 에 추가)
from __future__ import annotations
from datetime import datetime
from enum import Enum
from typing import Optional
from sqlmodel import SQLModel, Field
# from .models import now  (phase-2 정의 재사용)


class ConnectorDomain(str, Enum):
    mail = "mail"
    calendar = "calendar"
    chat = "chat"
    finance = "finance"
    health = "health"
    knowledge = "knowledge"


class ConnectorMode(str, Enum):
    mock = "mock"
    csv = "csv"
    healthkit = "healthkit"
    ics = "ics"
    real = "real"


class ConnState(str, Enum):
    connected = "connected"       # 정상 연결 (라이프 카드 on:true)
    disconnected = "disconnected" # 미연결 (life-data.js: "연결 안 됨")
    syncing = "syncing"           # 동기화 중
    error = "error"               # 마지막 sync 실패
    token_expired = "token_expired"  # 권한 만료 → 재인증 필요


class ConnectorAccount(SQLModel, table=True):
    """한 도메인의 한 제공자 계정. 예: mail/work(jiwoo@lumi.co), finance/woori."""
    __tablename__ = "connector_account"
    id: str = Field(primary_key=True)              # TEXT PK 예: "ca-mail-work"
    user_id: str = Field(default="jiwoo", foreign_key="person.id")  # 단일 데모 사용자(phase-15 멀티유저)
    domain: ConnectorDomain
    mode: ConnectorMode = ConnectorMode.mock
    provider: str = ""                             # "gmail" | "google_calendar" | "notion" | "woori" | "apple_health" ...
    external_account_id: str = ""                  # 제공자 계정 식별자(이메일 주소 등)
    name: str = ""                                 # 표시명 예: "회사" / "우리카드"
    kind: str = ""                                 # life-data.js 의 kind 예: "Lumi Workspace" / "신용/체크"
    tone: str = "ink"                              # tone 집합: blue|violet|coral|green|amber|ink|faint
    state: ConnState = ConnState.disconnected
    last_synced_at: Optional[datetime] = None
    last_label: str = "연결 안 됨"                  # 라이프 카드 last 문구 (예: "방금 동기화" / "오늘 09:12")
    token_enc: str = ""                            # 암호화된 토큰 JSON (crypto.py). 평문 저장 금지.
    scopes: str = ""                               # 콤마구분 OAuth 스코프
    error_detail: str = ""
    created_at: datetime = Field(default_factory=now)
    updated_at: datetime = Field(default_factory=now)


class ConnectorSyncState(SQLModel, table=True):
    """증분 sync 커서. 제공자별 델타 토큰 또는 since 타임스탬프."""
    __tablename__ = "connector_sync_state"
    id: str = Field(primary_key=True)              # "cs-<account_id>"
    account_id: str = Field(foreign_key="connector_account.id")
    cursor: str = ""                               # Gmail historyId / Google syncToken / ISO since
    page_token: str = ""                           # 진행 중 페이지네이션 토큰(중단 복구용)
    last_full_sync_at: Optional[datetime] = None
    last_delta_sync_at: Optional[datetime] = None
    items_seen: int = 0
    items_upserted: int = 0


class ExternalLink(SQLModel, table=True):
    """외부 제공자 객체 ↔ 내부 엔티티 매핑(중복/충돌 처리의 핵심).
       (account_id, external_id) UNIQUE 로 멱등 upsert."""
    __tablename__ = "external_link"
    id: str = Field(primary_key=True)              # "xl-<hash>"
    account_id: str = Field(foreign_key="connector_account.id", index=True)
    external_id: str = Field(index=True)           # 제공자측 고유 id (Gmail message id 등)
    entity_type: str                               # "email" | "event" | "finance_tx" | "health_sample" | "knowledge_item" | "chat_message"
    entity_id: str                                 # 내부 엔티티 PK (예: email.id)
    etag: str = ""                                 # 변경 감지용(있으면 비교 후 skip)
    external_updated_at: Optional[datetime] = None
    created_at: datetime = Field(default_factory=now)


class OAuthState(SQLModel, table=True):
    """OAuth Authorization Code 플로우의 CSRF state + PKCE verifier 임시 저장."""
    __tablename__ = "oauth_state"
    id: str = Field(primary_key=True)              # state nonce 자체를 PK 로
    domain: ConnectorDomain
    provider: str
    code_verifier: str = ""                        # PKCE
    redirect_after: str = "/life"                  # 콜백 후 프론트 복귀 경로
    created_at: datetime = Field(default_factory=now)


class ConnectorAccountLog(SQLModel, table=True):
    """연결/해제/동기화/오류 활동 로그(결재함 activity log 와 동일 톤)."""
    __tablename__ = "connector_account_log"
    id: str = Field(primary_key=True)              # "cl1", ...
    account_id: str = Field(foreign_key="connector_account.id")
    action: str                                    # "connected" | "disconnected" | "sync" | "error" | "token_refreshed" | "import"
    detail: str = ""
    items: int = 0
    created_at: datetime = Field(default_factory=now)
```

> `external_link` 의 `(account_id, external_id)` 는 **복합 UNIQUE 인덱스**로 만든다(Alembic 에서 `op.create_index(..., unique=True)`). 이것이 "같은 메일/일정을 두 번 받아도 한 번만 저장"(멱등성)을 보장하는 단일 지점이다.

#### Alembic 마이그레이션

```bash
cd /Users/kim1634/mycloud/personal/workspace/ari_assistant/workspace/backend
uv run alembic revision --autogenerate -m "phase13 connectors"
# 생성 파일에 external_link 복합 UNIQUE 인덱스 수동 보강:
#   op.create_index("ux_extlink_acct_ext", "external_link", ["account_id", "external_id"], unique=True)
uv run alembic upgrade head
# 왕복:
uv run alembic downgrade -1 && uv run alembic upgrade head
```

#### 시드 — mock 계정(라이프/메일 카드 값 그대로)

`run_seed()` 의 하위 헬퍼 `_seed_connectors(session)` 를 추가한다(phase-2 의 `_seed_dashboard` 와 동일 패턴 — `run_seed` 내부에서만 호출). 값은 **원본 `life-data.js: sources` 와 `mail-data.js: accounts` 그대로** 이식한다.

```python
# backend/app/seed.py  (run_seed 안에서 _seed_connectors(s) 호출)
# (domain, id, provider, name, kind, tone, on, last)  ← life-data.js sources / mail-data.js accounts 그대로
CONNECTORS = [
    # mail (mail-data.js accounts) — 회사/개인/사이드 3계정
    ("mail", "ca-mail-work",     "google_workspace", "회사",  "Lumi Workspace", "coral",  True, "방금 동기화"),
    ("mail", "ca-mail-personal", "gmail",            "개인",  "Google",         "blue",   True, "방금 동기화"),
    ("mail", "ca-mail-side",     "hey",              "사이드","HEY",            "violet", True, "방금 동기화"),
    # calendar (cal-data.js cals 는 카테고리지만 연동 단위는 google_calendar 1계정으로 묶음)
    ("calendar", "ca-cal-google", "google_calendar", "Google 캘린더", "캘린더", "blue", True, "방금 동기화"),
    # health (life-data.js sources.health)
    ("health", "ca-health-apple", "apple_health", "Apple Health", "건강 앱",  "coral",  True,  "방금 동기화"),
    ("health", "ca-health-watch", "apple_watch",  "Apple Watch",  "웨어러블", "violet", True,  "1분 전"),
    ("health", "ca-health-fit",   "google_fit",   "Google Fit",   "활동 추적","green",  False, "연결 안 됨"),
    # finance (life-data.js sources.finance)
    ("finance", "ca-fin-woori", "woori",      "우리카드",   "신용/체크", "blue",  True,  "오늘 09:12"),
    ("finance", "ca-fin-toss",  "toss",       "토스",       "계좌 통합", "coral", True,  "오늘 08:40"),
    ("finance", "ca-fin-kakao", "kakaobank",  "카카오뱅크", "입출금",    "amber", True,  "어제"),
    ("finance", "ca-fin-kb",    "kb_sec",     "KB증권",     "투자",      "green", False, "연결 안 됨"),
    # knowledge (life-data.js sources.knowledge)
    ("knowledge", "ca-kn-notion",  "notion",   "Notion",    "메모·문서",     "violet", True, "오늘"),
    ("knowledge", "ca-kn-web",     "clipper",  "웹 클리퍼", "아티클 스크랩", "blue",   True, "2시간 전"),
    ("knowledge", "ca-kn-readwise","readwise", "Readwise",  "하이라이트",    "coral",  True, "어제"),
]

def _seed_connectors(s):
    from .models import ConnectorAccount, ConnState, ConnectorMode
    for domain, cid, provider, name, kind, tone, on, last in CONNECTORS:
        s.add(ConnectorAccount(
            id=cid, user_id="jiwoo", domain=domain, mode=ConnectorMode.mock,
            provider=provider, name=name, kind=kind, tone=tone,
            external_account_id=name, last_label=last,
            state=ConnState.connected if on else ConnState.disconnected,
        ))
```

> 시드는 **mock 모드** 의 계정이다. `on:false` 인 Google Fit·KB증권은 `disconnected` 로 — 원본 `life-data.js` 의 `"연결 안 됨"` 문구(`last`)를 `last_label` 에 그대로 보존한다. 이메일 계정은 `mail-data.js` 의 `id`(work/personal/side)·`name`(회사/개인/사이드)·`kind`(Lumi Workspace/Google/HEY)·`tone`(coral/blue/violet) 을 1바이트도 안 바꾸고 옮긴다.

---

### 3.2 토큰 암호화 — `crypto.py`

평문 토큰은 절대 DB 에 저장하지 않는다. `ARI_SECRET_KEY` 에서 Fernet 키를 파생해 토큰 JSON 을 대칭 암호화한다.

```python
# backend/app/crypto.py
import base64
import hashlib
import json
from cryptography.fernet import Fernet, InvalidToken
from .config import get_settings


def _fernet() -> Fernet:
    # ARI_SECRET_KEY -> 32바이트 -> urlsafe base64 -> Fernet 키
    raw = get_settings().ari_secret_key.encode("utf-8")
    key = base64.urlsafe_b64encode(hashlib.sha256(raw).digest())
    return Fernet(key)


def encrypt_token(token_dict: dict) -> str:
    """{'access_token':..., 'refresh_token':..., 'expires_at':...} -> 암호문 str."""
    return _fernet().encrypt(json.dumps(token_dict).encode("utf-8")).decode("utf-8")


def decrypt_token(enc: str) -> dict:
    if not enc:
        return {}
    try:
        return json.loads(_fernet().decrypt(enc.encode("utf-8")).decode("utf-8"))
    except (InvalidToken, ValueError):
        return {}   # 키 회전/손상 → 빈 dict (호출측은 token_expired 처리)
```

> 운영(phase-15)에서는 `ARI_SECRET_KEY` 를 시크릿 매니저에서 주입하고 키 회전 정책을 둔다. 본 phase 는 env 주입 + 손상 시 안전 폴백(`{}` → 재인증)만 보장.

---

### 3.3 Base 인터페이스 — `connectors/base.py`

CONTRACT 의 4개 메서드 `fetch / sync / normalize / write` 를 추상화한다. LLM `LLMProvider`(ABC) 패턴을 그대로 따른다.

```python
# backend/app/connectors/base.py
from __future__ import annotations
from abc import ABC, abstractmethod
from dataclasses import dataclass, field
from datetime import datetime, timezone
from typing import Any, Iterable
from sqlmodel import Session, select
from ..models import ConnectorAccount, ConnectorSyncState, ExternalLink, ConnState
import hashlib


def _now():
    return datetime.now(timezone.utc)


@dataclass
class RawRecord:
    """제공자에서 fetch 한 1건의 원시 레코드(provider 페이로드)."""
    external_id: str
    payload: dict
    etag: str = ""
    external_updated_at: datetime | None = None


@dataclass
class NormalizedRecord:
    """normalize() 산출물 — 내부 모델로 write 가능한 정규화 dict."""
    entity_type: str                 # "email" | "event" | "finance_tx" | ...
    external_id: str
    fields: dict[str, Any] = field(default_factory=dict)
    etag: str = ""
    external_updated_at: datetime | None = None


@dataclass
class SyncResult:
    domain: str
    account_id: str
    seen: int = 0
    upserted: int = 0
    skipped: int = 0
    errors: int = 0
    events_published: list[str] = field(default_factory=list)
    detail: str = ""


class BaseConnector(ABC):
    """모든 도메인 커넥터의 공통 계약. Mock/Real 이 동일하게 구현한다(인터페이스 불변)."""
    domain: str = "base"
    entity_type: str = "record"

    def __init__(self, account: ConnectorAccount):
        self.account = account

    # ── (1) fetch: 제공자에서 원시 레코드를 가져온다(증분 since/cursor) ──
    @abstractmethod
    def fetch(self, session: Session, *, full: bool = False) -> Iterable[RawRecord]: ...

    # ── (2) normalize: provider 페이로드 → 내부 모델 dict ──
    @abstractmethod
    def normalize(self, raw: RawRecord) -> NormalizedRecord: ...

    # ── (3) write: 정규화 레코드를 내부 테이블에 upsert(external_link 멱등) ──
    @abstractmethod
    def write(self, session: Session, norm: NormalizedRecord) -> tuple[str, bool]:
        """return (entity_id, created)  — created=False 면 update/skip."""
        ...

    # ── (4) sync: fetch→normalize→write 오케스트레이션 + event_bus 발행 ──
    def sync(self, session: Session, *, full: bool = False) -> SyncResult:
        from ..automation.event_bus import bus     # phase-7 정본 (지연 import 로 순환 방지)
        res = SyncResult(domain=self.domain, account_id=self.account.id)
        self._set_state(session, ConnState.syncing)
        try:
            for raw in self.fetch(session, full=full):
                res.seen += 1
                if self._is_duplicate(session, raw):      # etag 동일 → skip(멱등)
                    res.skipped += 1
                    continue
                norm = self.normalize(raw)
                entity_id, created = self.write(session, norm)
                self._upsert_link(session, raw, norm, entity_id)
                if created:
                    res.upserted += 1
                    evt = self.event_for(norm, entity_id)  # 예: "mail.received"
                    if evt:
                        bus.publish(evt, {"account_id": self.account.id,
                                          "entity_type": norm.entity_type,
                                          "entity_id": entity_id})
                        res.events_published.append(evt)
            session.commit()
            self._finish_ok(session, res)
        except Exception as e:               # 부분 실패도 서버 죽지 않음(오프라인 폴백 철학)
            session.rollback()
            res.errors += 1
            res.detail = f"{type(e).__name__}: {e}"
            self._finish_error(session, res.detail)
        return res

    # ── 도메인이 발행할 연합 이벤트 타입(없으면 None) ──
    def event_for(self, norm: NormalizedRecord, entity_id: str) -> str | None:
        return None

    # ── 공통: 중복/멱등 ──
    def _is_duplicate(self, session: Session, raw: RawRecord) -> bool:
        link = session.exec(
            select(ExternalLink).where(ExternalLink.account_id == self.account.id,
                                       ExternalLink.external_id == raw.external_id)).first()
        return bool(link and raw.etag and link.etag == raw.etag)

    def _upsert_link(self, session, raw, norm, entity_id):
        link = session.exec(
            select(ExternalLink).where(ExternalLink.account_id == self.account.id,
                                       ExternalLink.external_id == raw.external_id)).first()
        if link:
            link.etag = norm.etag; link.entity_id = entity_id
            link.external_updated_at = norm.external_updated_at
        else:
            xlid = "xl-" + hashlib.sha1(f"{self.account.id}:{raw.external_id}".encode()).hexdigest()[:12]
            session.add(ExternalLink(id=xlid, account_id=self.account.id,
                                     external_id=raw.external_id, entity_type=norm.entity_type,
                                     entity_id=entity_id, etag=norm.etag,
                                     external_updated_at=norm.external_updated_at))

    # ── 상태/커서/로그 ──
    def _state_row(self, session) -> ConnectorSyncState:
        st = session.get(ConnectorSyncState, f"cs-{self.account.id}")
        if not st:
            st = ConnectorSyncState(id=f"cs-{self.account.id}", account_id=self.account.id)
            session.add(st)
        return st

    def _set_state(self, session, state: ConnState):
        self.account.state = state; self.account.updated_at = _now()
        session.add(self.account); session.commit()

    def _finish_ok(self, session, res: SyncResult):
        self.account.state = ConnState.connected
        self.account.last_synced_at = _now()
        self.account.last_label = "방금 동기화"
        self.account.error_detail = ""
        st = self._state_row(session)
        st.items_seen += res.seen; st.items_upserted += res.upserted
        st.last_delta_sync_at = _now()
        session.add_all([self.account, st]); session.commit()
        self._log(session, "sync", f"upserted={res.upserted} skipped={res.skipped}", res.upserted)

    def _finish_error(self, session, detail: str):
        self.account.state = ConnState.error
        self.account.error_detail = detail
        session.add(self.account); session.commit()
        self._log(session, "error", detail)

    def _log(self, session, action: str, detail: str = "", items: int = 0):
        from ..models import ConnectorAccountLog
        import uuid
        session.add(ConnectorAccountLog(id="cl-" + uuid.uuid4().hex[:8],
                    account_id=self.account.id, action=action, detail=detail, items=items))
        session.commit()
```

> `write()` 는 `(entity_id, created)` 를 반환한다 — `created=True` 일 때만 이벤트를 발행해 **재동기화로 같은 메일이 다시 알림을 만드는 일**을 막는다. `_is_duplicate` 는 etag 동일 시 skip — 멱등성의 두 번째 방어선(첫 번째는 `external_link` UNIQUE).

---

### 3.4 레지스트리 — `connectors/registry.py`

`CONNECTOR_<DOMAIN>` 로 mock/real 을 고르는 단일 지점. 페이지/라우터는 **항상 레지스트리를 통해서만** 커넥터를 얻는다(코드 불변의 핵심).

```python
# backend/app/connectors/registry.py
from __future__ import annotations
from sqlmodel import Session, select
from ..config import get_settings
from ..models import ConnectorAccount, ConnectorDomain
from .base import BaseConnector

# 도메인 × 모드 → 구현 클래스 (lazy import 로 의존성 격리)
def _impl(domain: str, mode: str):
    if domain == "mail":
        from .mail.mock import MockMailConnector
        from .mail.real_gmail import GmailConnector
        from .mail.real_imap import ImapConnector
        return {"mock": MockMailConnector, "real": GmailConnector, "imap": ImapConnector}.get(mode, MockMailConnector)
    if domain == "calendar":
        from .calendar.mock import MockCalendarConnector
        from .calendar.real_google import GoogleCalendarConnector
        from .calendar.ics_import import IcsConnector
        return {"mock": MockCalendarConnector, "real": GoogleCalendarConnector, "ics": IcsConnector}.get(mode, MockCalendarConnector)
    if domain == "finance":
        from .finance.mock import MockFinanceConnector
        from .finance.csv_import import CsvFinanceConnector
        from .finance.real_aggregator import AggregatorConnector
        return {"mock": MockFinanceConnector, "csv": CsvFinanceConnector, "real": AggregatorConnector}.get(mode, MockFinanceConnector)
    if domain == "health":
        from .health.mock import MockHealthConnector
        from .health.healthkit_import import HealthKitConnector
        from .health.real_fit import GoogleFitConnector
        return {"mock": MockHealthConnector, "healthkit": HealthKitConnector, "real": GoogleFitConnector}.get(mode, MockHealthConnector)
    if domain == "knowledge":
        from .knowledge.mock import MockKnowledgeConnector
        from .knowledge.real_notion import NotionConnector
        return {"mock": MockKnowledgeConnector, "real": NotionConnector}.get(mode, MockKnowledgeConnector)
    if domain == "chat":
        from .chat.mock import MockChatConnector
        from .chat.real_webhook import WebhookChatConnector
        return {"mock": MockChatConnector, "real": WebhookChatConnector}.get(mode, MockChatConnector)
    raise ValueError(f"unknown domain {domain}")


class ConnectorRegistry:
    @staticmethod
    def mode_for(domain: str) -> str:
        return getattr(get_settings(), f"connector_{domain}", "mock")

    @staticmethod
    def get(session: Session, account: ConnectorAccount) -> BaseConnector:
        """계정별 커넥터 인스턴스. 모드는 env(CONNECTOR_<DOMAIN>) > 계정.mode 순.
        단, mock 시드 계정은 env 와 무관하게 mock 으로(데모 결정성)."""
        env_mode = ConnectorRegistry.mode_for(account.domain)
        mode = env_mode if account.mode != "mock" or env_mode != "real" else "mock"
        # 정책: env=real 이면 real 계정만 real 로. 시드 mock 계정은 그대로 mock.
        mode = env_mode if account.mode.value != "mock" else (
            "mock" if env_mode == "real" else env_mode)
        Impl = _impl(account.domain, account.mode.value if account.mode.value != "mock" else env_mode)
        return Impl(account)

    @staticmethod
    def accounts(session: Session, domain: str | None = None) -> list[ConnectorAccount]:
        q = select(ConnectorAccount)
        if domain:
            q = q.where(ConnectorAccount.domain == domain)
        return session.exec(q.order_by(ConnectorAccount.id)).all()
```

> **모드 결정 정본**: 계정 row 의 `mode` 가 우선(연결 시 real 로 승격된 계정은 `mode=real`). env `CONNECTOR_<DOMAIN>=real` 은 "이 도메인에서 real 연동을 허용한다"는 게이트. 시드 mock 계정(`mode=mock`)은 env 가 real 이어도 **mock 으로 남아** 데모/CI 결정성을 지킨다 — real 데이터는 OAuth 로 새로 연결한 계정(`mode=real`)에서만 들어온다. 따라서 **mock 과 real 이 공존**할 수 있다(예: 메일은 real, 금융은 mock).

---

### 3.5 도메인 구현 — mail (대표 사례, 가장 상세)

#### 3.5.1 `mail/mock.py` — 시드 기반 (기본값)

`mail-data.js: emails` 를 그대로 fetch 한다. 이미 phase-9 가 `email` 테이블에 시드했다면 mock 은 "no-op sync"(이미 있는 데이터)로 동작하되, 데모용 "새 메일 1건 도착" 시뮬레이션을 위한 `simulate_new()` 진입점을 둔다.

```python
# backend/app/connectors/mail/mock.py
from datetime import datetime, timezone
from sqlmodel import Session
from ..base import BaseConnector, RawRecord, NormalizedRecord
from .normalize import normalize_email

# mail-data.js emails 에서 가져온 "새로 도착할" 시뮬레이션 메일(연합 데모용)
_SIM = [{
    "id": "sim-m1", "account": "work", "from": "hyunwoo", "to": "나",
    "subject": "온보딩 시안 v3 — 금요일까지 피드백 부탁드려요",
    "date": "오늘 09:12", "read": False, "starred": False,
    "preview": "지우님, 리뷰에서 나온 코멘트 반영해서 v3 정리했어요…",
    "labels": ["디자인"],
}]


class MockMailConnector(BaseConnector):
    domain = "mail"
    entity_type = "email"

    def fetch(self, session: Session, *, full: bool = False):
        # mock: 기본은 빈 스트림(시드는 이미 적재됨). 데모 시 simulate 로 1건 흘림.
        src = getattr(self, "_sim_payloads", [])
        for p in src:
            yield RawRecord(external_id=p["id"], payload=p,
                            etag=p.get("date", ""),
                            external_updated_at=datetime.now(timezone.utc))

    def simulate_new(self):
        """데모: '새 메일 도착' 1건을 다음 sync 에 흘린다(연합 트리거)."""
        self._sim_payloads = _SIM
        return self

    def normalize(self, raw: RawRecord) -> NormalizedRecord:
        return normalize_email(self.account, raw)   # provider-무관 정규화 공유

    def write(self, session: Session, norm: NormalizedRecord):
        return _upsert_email(session, norm)

    def event_for(self, norm, entity_id):
        return "mail.received"
```

#### 3.5.2 `mail/real_gmail.py` — Gmail API

OAuth2 액세스 토큰으로 Gmail REST(`users.messages.list` + `users.messages.get`)를 호출하고, 증분은 `historyId`(델타) 또는 `q=after:<epoch>` 로 처리한다.

```python
# backend/app/connectors/mail/real_gmail.py
import base64
import httpx
from datetime import datetime, timezone
from sqlmodel import Session
from ..base import BaseConnector, RawRecord, NormalizedRecord
from ..oauth import valid_access_token        # 만료 시 자동 refresh
from ..ratelimit import RateLimiter
from .normalize import normalize_email
from ...config import get_settings

API = "https://gmail.googleapis.com/gmail/v1"


class GmailConnector(BaseConnector):
    domain = "mail"
    entity_type = "email"
    _rl = RateLimiter(rate=4, per=1.0)   # Gmail per-user QPS 보수적 제한

    def fetch(self, session: Session, *, full: bool = False):
        st = self._state_row(session)
        token = valid_access_token(session, self.account)   # token_expired 면 예외 → sync 가 error 처리
        headers = {"Authorization": f"Bearer {token}"}
        timeout = get_settings().connector_http_timeout
        page_size = get_settings().sync_page_size
        params = {"maxResults": page_size, "q": "" if full else "newer_than:7d"}
        if st.cursor and not full:
            params["q"] = f"after:{st.cursor}"              # epoch seconds 커서
        with httpx.Client(timeout=timeout) as c:
            self._rl.acquire()
            r = c.get(f"{API}/users/me/messages", headers=headers, params=params)
            self._rl.handle_response(r)                     # 429 → 백오프
            r.raise_for_status()
            for m in r.json().get("messages", []):
                self._rl.acquire()
                d = c.get(f"{API}/users/me/messages/{m['id']}", headers=headers,
                          params={"format": "full"})
                self._rl.handle_response(d); d.raise_for_status()
                msg = d.json()
                yield RawRecord(external_id=msg["id"], payload=msg,
                                etag=msg.get("historyId", ""),
                                external_updated_at=_epoch_to_dt(msg.get("internalDate")))
        st.cursor = str(int(datetime.now(timezone.utc).timestamp()))
        session.add(st); session.commit()

    def normalize(self, raw: RawRecord) -> NormalizedRecord:
        return normalize_email(self.account, raw, provider="gmail")

    def write(self, session: Session, norm: NormalizedRecord):
        return _upsert_email(session, norm)

    def event_for(self, norm, entity_id):
        return "mail.received"


def _epoch_to_dt(ms: str | None):
    return datetime.fromtimestamp(int(ms) / 1000, tz=timezone.utc) if ms else None
```

`mail/real_imap.py`(HEY/일반 IMAP)는 `imaplib` 로 `UID SEARCH SINCE` 증분을 구현하되 동일 `RawRecord`/`normalize_email` 로 귀결한다(로컬 우선 대안 — OAuth 불필요).

#### 3.5.3 `mail/normalize.py` — provider → 내부 `email` 모델 (정규화 골든)

**provider 가 무엇이든** 동일한 내부 `email` 필드로 매핑한다. phase-9 의 `email` 모델 필드(`account/from/to/subject/body/labels/read/starred/attachments + ai{...}`)를 타깃으로 한다. 내부 `ai` 묶음(summary/priority/category/tasks/events/replies/file)은 **로컬 LLM(phase-9 의 메일 분석)** 이 채운다 — provider 는 raw 만 준다.

```python
# backend/app/connectors/mail/normalize.py
from datetime import datetime, timezone
from ..base import RawRecord, NormalizedRecord
from ...models import ConnectorAccount

# mail-data.js account id ↔ connector_account 매핑(시드 정합)
_ACCOUNT_KEY = {"ca-mail-work": "work", "ca-mail-personal": "personal", "ca-mail-side": "side"}


def normalize_email(account: ConnectorAccount, raw: RawRecord, provider: str = "mock") -> NormalizedRecord:
    p = raw.payload
    if provider == "gmail":
        headers = {h["name"].lower(): h["value"]
                   for h in p.get("payload", {}).get("headers", [])}
        fields = {
            "account": _ACCOUNT_KEY.get(account.id, "work"),
            "from_addr": headers.get("from", ""),
            "to_addr": headers.get("to", "나"),
            "subject": headers.get("subject", "(제목 없음)"),
            "body": _gmail_body(p),
            "labels": _gmail_labels(p.get("labelIds", [])),
            "read": "UNREAD" not in p.get("labelIds", []),
            "starred": "STARRED" in p.get("labelIds", []),
            "has_attach": _gmail_has_attach(p),
            "received_at": _internal_dt(p.get("internalDate")),
            "ai": None,          # phase-9 메일 분석(로컬 LLM)이 후처리로 채움
        }
    else:  # mock(mail-data.js shape) / imap 등은 이미 내부 키와 유사
        fields = {
            "account": p.get("account", "work"),
            "from_addr": p.get("from", ""), "to_addr": p.get("to", "나"),
            "subject": p.get("subject", ""), "body": "\n".join(p.get("body", [])) if isinstance(p.get("body"), list) else p.get("body", ""),
            "labels": p.get("labels", []), "read": p.get("read", False),
            "starred": p.get("starred", False), "has_attach": p.get("hasAttach", False),
            "received_at": datetime.now(timezone.utc), "ai": p.get("ai"),
        }
    return NormalizedRecord(entity_type="email", external_id=raw.external_id,
                            fields=fields, etag=raw.etag,
                            external_updated_at=raw.external_updated_at)

def _internal_dt(ms):
    return datetime.fromtimestamp(int(ms) / 1000, tz=timezone.utc) if ms else datetime.now(timezone.utc)
# _gmail_body / _gmail_labels / _gmail_has_attach: MIME 파트 파싱(생략 — base64url 디코딩, 라벨 한국어 매핑)
```

> **정규화 골든의 의미**: `test_normalize_golden.py` 가 "Gmail 페이로드 픽스처 → 위 `fields` dict" 의 매핑을 **고정**한다. provider 가 바뀌어도(IMAP/HEY) 동일 `fields` 로 귀결되어야 페이지(메일/알림/작업)가 코드 변경 없이 동작한다. 라벨은 Gmail `CATEGORY_PROMOTIONS` 등 → 내부 한국어 라벨(`["디자인"]`, `["금융"]` 등 `mail-data.js` 톤)로 매핑한다.

#### 3.5.4 공통 upsert — `email` 멱등 저장

```python
# backend/app/connectors/mail/_upsert.py (또는 normalize.py 하단)
import uuid
from sqlmodel import Session, select
from ...models import Email, ExternalLink     # Email = phase-9 모델

def _upsert_email(session: Session, norm) -> tuple[str, bool]:
    link = session.exec(select(ExternalLink).where(
        ExternalLink.external_id == norm.external_id,
        ExternalLink.entity_type == "email")).first()
    if link:                                   # 이미 존재 → update(멱등)
        e = session.get(Email, link.entity_id)
        if e:
            for k, v in norm.fields.items():
                if v is not None and hasattr(e, k):
                    setattr(e, k, v)
            session.add(e)
            return e.id, False
    eid = "m-" + uuid.uuid4().hex[:8]
    e = Email(id=eid, **{k: v for k, v in norm.fields.items() if v is not None})
    session.add(e)
    return eid, True
```

> 다른 도메인(calendar/finance/health/knowledge/chat)도 **정확히 같은 4단 구조**(mock.py / real_*.py / normalize.py / _upsert)를 따른다. 표는 §3.6.

---

### 3.6 도메인별 매핑 요약표

| 도메인 | mock 출처(시드) | real 제공자 | 로컬 우선 대안 | 정규화 타깃(내부 모델, phase) | 발행 이벤트 |
|---|---|---|---|---|---|
| **mail** | `mail-data.js: emails/accounts` | Gmail API, Google Workspace | IMAP(HEY/일반) | `email`(phase-9) | `mail.received` |
| **calendar** | `cal-data.js: events/focusBlocks` | Google Calendar API | `.ics` 파일 임포트 | `event`/`focus_block`/`meeting`(phase-8) | `calendar.updated` |
| **chat** | (phase-9 알림 시드) | 메신저 Webhook/API | — | `notification`/`chat_message`(phase-9) | `chat.message` |
| **finance** | `life-data.js: finance` | 카드/뱅크 아그리게이터 | **CSV 임포트**(카드 명세) | `finance`/`finance_tx`(phase-11) | `finance.transaction` |
| **health** | `life-data.js: health` | Google Fit API | **HealthKit export(.zip/.xml)** | `health`(rings/vitals/sleep)(phase-11) | `health.sample` |
| **knowledge** | `life-data.js: knowledge` | Notion API, Readwise | 웹 클리퍼, Markdown 노트 | `knowledge_item` + RAG ingest(phase-10/11) | `knowledge.ingested` |

도메인 구현 디테일 메모:
- **calendar**: Google Calendar 증분은 `syncToken`(델타) — `events.list?syncToken=...`, 410 GONE 시 full resync 후 새 토큰 획득. `.ics` 임포트(`ics_import.py`)는 `icalendar` 파싱 → 동일 `event` 정규화(로컬 우선, OAuth 불필요). 정규화는 `cal-data.js` 의 `{day,start,end,title,cal,loc,people,note,soon}` 키로 귀결.
- **finance**: 한국 금융 OAuth 아그리게이터는 불안정/지역 제약 → **CSV 임포트를 1급 시민**으로(`csv_import.py`). 우리카드/토스 CSV 헤더 → `finance_tx{date, merchant, amount, category, account}`. 카테고리는 `life-data.js: finance.cats`(식비/쇼핑/교통/카페·간식/구독/문화·여가) 로 매핑. 금액 단위는 원본대로 정수 원화.
- **health**: Apple Health 는 공식 API 가 없어 **HealthKit export(.xml)** 임포트가 정본(`healthkit_import.py`) — `<Record type="HKQuantityTypeIdentifierStepCount" .../>` 파싱 → `health.rings`(걸음/운동/수면). Google Fit 은 OAuth real. 원본 `life-data.js` 의 단위·문구("7,240보", "7h 12m", "안정 심박 62 bpm")를 표시 포맷으로 보존.
- **knowledge**: Notion/Readwise/클리퍼 → `knowledge_item{type: article|note|idea|highlight, title, src, excerpt, tags, ai}` (life-data.js shape) + **RAG ingest 트리거**(phase-10 `backend/app/rag/ingest`). knowledge_item 은 리서치(phase-10)와 라이프(phase-11)가 공유.
- **chat**: 인바운드 Webhook(`real_webhook.py`)으로 메신저 메시지 수신 → `notification` 트리아지(phase-9)로 흘림. OAuth 대신 서명 검증.

---

### 3.7 OAuth2 공통 헬퍼 — `connectors/oauth.py`

Authorization Code + PKCE. start(인증 URL 생성·state 저장) → callback(code→token 교환·암호화 저장) → refresh(만료 시 자동).

```python
# backend/app/connectors/oauth.py
import base64, hashlib, os, secrets, time
import httpx
from sqlmodel import Session
from ..config import get_settings
from ..crypto import encrypt_token, decrypt_token
from ..models import ConnectorAccount, OAuthState, ConnState

PROVIDERS = {
    "google": {  # Gmail + Google Calendar
        "auth": "https://accounts.google.com/o/oauth2/v2/auth",
        "token": "https://oauth2.googleapis.com/token",
        "scopes": {"gmail": ["https://www.googleapis.com/auth/gmail.readonly"],
                   "google_calendar": ["https://www.googleapis.com/auth/calendar.events"]},
    },
    "notion": {"auth": "https://api.notion.com/v1/oauth/authorize",
               "token": "https://api.notion.com/v1/oauth/token", "scopes": {"notion": []}},
}


def start_oauth(session: Session, domain: str, provider: str, redirect_after: str = "/life") -> str:
    st = get_settings()
    fam = "google" if provider in ("gmail", "google_calendar") else provider
    conf = PROVIDERS[fam]
    state = secrets.token_urlsafe(24)
    verifier = base64.urlsafe_b64encode(os.urandom(40)).decode().rstrip("=")
    challenge = base64.urlsafe_b64encode(hashlib.sha256(verifier.encode()).digest()).decode().rstrip("=")
    session.add(OAuthState(id=state, domain=domain, provider=provider,
                           code_verifier=verifier, redirect_after=redirect_after))
    session.commit()
    scopes = " ".join(conf["scopes"].get(provider, []))
    params = {"client_id": st.google_client_id if fam == "google" else st.notion_client_id,
              "redirect_uri": st.google_redirect_uri, "response_type": "code",
              "scope": scopes, "state": state, "access_type": "offline", "prompt": "consent",
              "code_challenge": challenge, "code_challenge_method": "S256"}
    return f"{conf['auth']}?" + "&".join(f"{k}={httpx.QueryParams({k:v})[k]}" for k, v in params.items())


def finish_oauth(session: Session, code: str, state: str) -> ConnectorAccount:
    os_row = session.get(OAuthState, state)
    if not os_row:
        raise ValueError("invalid oauth state")
    st = get_settings()
    fam = "google" if os_row.provider in ("gmail", "google_calendar") else os_row.provider
    conf = PROVIDERS[fam]
    data = {"code": code, "grant_type": "authorization_code",
            "redirect_uri": st.google_redirect_uri, "code_verifier": os_row.code_verifier,
            "client_id": st.google_client_id if fam == "google" else st.notion_client_id,
            "client_secret": st.google_client_secret if fam == "google" else st.notion_client_secret}
    r = httpx.post(conf["token"], data=data, timeout=st.connector_http_timeout)
    r.raise_for_status()
    tok = r.json()
    tok["expires_at"] = int(time.time()) + int(tok.get("expires_in", 3600))
    # 새 real 계정 생성(또는 기존 계정 승격) → mode=real
    acct = _ensure_account(session, os_row.domain, os_row.provider)
    acct.mode = "real"; acct.state = ConnState.connected
    acct.token_enc = encrypt_token(tok); acct.scopes = tok.get("scope", "")
    acct.last_label = "방금 연결됨"
    session.add(acct); session.delete(os_row); session.commit()
    return acct


def valid_access_token(session: Session, account: ConnectorAccount) -> str:
    """만료되면 refresh_token 으로 갱신. 갱신 불가면 token_expired 로 표시 후 예외."""
    tok = decrypt_token(account.token_enc)
    if not tok:
        account.state = ConnState.token_expired; session.add(account); session.commit()
        raise PermissionError("token missing/corrupt")
    if tok.get("expires_at", 0) > int(time.time()) + 60:
        return tok["access_token"]
    # refresh
    st = get_settings()
    fam = "google" if account.provider in ("gmail", "google_calendar") else account.provider
    conf = PROVIDERS[fam]
    rt = tok.get("refresh_token")
    if not rt:
        account.state = ConnState.token_expired; session.add(account); session.commit()
        raise PermissionError("no refresh_token")
    r = httpx.post(conf["token"], data={
        "grant_type": "refresh_token", "refresh_token": rt,
        "client_id": st.google_client_id if fam == "google" else st.notion_client_id,
        "client_secret": st.google_client_secret if fam == "google" else st.notion_client_secret},
        timeout=st.connector_http_timeout)
    if r.status_code != 200:
        account.state = ConnState.token_expired; session.add(account); session.commit()
        raise PermissionError("refresh failed")
    new = r.json(); new["refresh_token"] = rt
    new["expires_at"] = int(time.time()) + int(new.get("expires_in", 3600))
    account.token_enc = encrypt_token(new); session.add(account); session.commit()
    return new["access_token"]
```

---

### 3.8 레이트리밋 — `connectors/ratelimit.py`

```python
# backend/app/connectors/ratelimit.py
import threading, time
import httpx


class RateLimiter:
    """간단 토큰 버킷 + 429/Retry-After 지수 백오프."""
    def __init__(self, rate: float = 4, per: float = 1.0, max_backoff: float = 32.0):
        self.rate, self.per = rate, per
        self.allowance = rate; self.last = time.monotonic()
        self.max_backoff = max_backoff; self._lock = threading.Lock()

    def acquire(self):
        with self._lock:
            now = time.monotonic()
            self.allowance += (now - self.last) * (self.rate / self.per); self.last = now
            if self.allowance > self.rate: self.allowance = self.rate
            if self.allowance < 1.0:
                time.sleep((1.0 - self.allowance) * (self.per / self.rate)); self.allowance = 0.0
            else:
                self.allowance -= 1.0

    def handle_response(self, r: httpx.Response, attempt: int = 0):
        if r.status_code == 429:
            retry = float(r.headers.get("Retry-After", min(2 ** attempt, self.max_backoff)))
            time.sleep(min(retry, self.max_backoff))
            return True   # 호출측이 재시도
        return False
```

---

### 3.9 라우터 — `routers/connectors.py`

phase-2 규약: 라우터 내부 prefix 없음, `main.py` 에서 `include_router(connectors.router, prefix="/api", tags=["connectors"])`.

```python
# backend/app/routers/connectors.py
from fastapi import APIRouter, Depends, HTTPException, UploadFile, File
from fastapi.responses import RedirectResponse
from sqlmodel import Session, select
from ..db import get_session
from ..models import ConnectorAccount, ConnectorAccountLog, ConnState
from ..connectors.registry import ConnectorRegistry
from ..connectors import oauth as oauthlib
from ..schemas import (ConnectorStatusOut, OAuthStartOut, SyncResultOut, ImportResultOut)
from ..config import get_settings

router = APIRouter()   # prefix 없음. main.py 에서 prefix="/api".


@router.get("/connectors", response_model=list[ConnectorStatusOut])
def list_connectors(domain: str | None = None, s: Session = Depends(get_session)):
    rows = ConnectorRegistry.accounts(s, domain)
    return [ConnectorStatusOut.from_account(a, ConnectorRegistry.mode_for(a.domain)) for a in rows]


@router.post("/connectors/{account_id}/sync", response_model=SyncResultOut)
def sync_one(account_id: str, full: bool = False, s: Session = Depends(get_session)):
    a = s.get(ConnectorAccount, account_id)
    if not a: raise HTTPException(404, "connector account not found")
    conn = ConnectorRegistry.get(s, a)
    res = conn.sync(s, full=full)
    return SyncResultOut(**res.__dict__)


@router.post("/connectors/sync-all", response_model=list[SyncResultOut])
def sync_all(s: Session = Depends(get_session)):
    out = []
    for a in ConnectorRegistry.accounts(s):
        if a.state == ConnState.disconnected:    # 연결 안 된 계정은 건너뜀
            continue
        out.append(SyncResultOut(**ConnectorRegistry.get(s, a).sync(s).__dict__))
    return out


# ── OAuth ──
@router.get("/connectors/oauth/start", response_model=OAuthStartOut)
def oauth_start(domain: str, provider: str, redirect_after: str = "/life",
                s: Session = Depends(get_session)):
    if not get_settings().google_client_id and provider in ("gmail", "google_calendar"):
        raise HTTPException(400, "google_client_id 미설정 — real 연동 비활성")
    url = oauthlib.start_oauth(s, domain, provider, redirect_after)
    return OAuthStartOut(authorize_url=url)

@router.get("/connectors/oauth/callback")
def oauth_callback(code: str, state: str, s: Session = Depends(get_session)):
    try:
        acct = oauthlib.finish_oauth(s, code, state)
    except Exception as e:
        return RedirectResponse(url=f"{get_settings().frontend_origin}/life?connect=error")
    # 연결 직후 1회 초기 sync
    ConnectorRegistry.get(s, acct).sync(s, full=True)
    redirect = s and "/life"
    return RedirectResponse(url=f"{get_settings().frontend_origin}/life?connect=ok&domain={acct.domain}")


@router.post("/connectors/{account_id}/disconnect", response_model=ConnectorStatusOut)
def disconnect(account_id: str, s: Session = Depends(get_session)):
    a = s.get(ConnectorAccount, account_id)
    if not a: raise HTTPException(404, "not found")
    a.state = ConnState.disconnected; a.token_enc = ""; a.last_label = "연결 안 됨"
    import uuid
    s.add(a); s.add(ConnectorAccountLog(id="cl-"+uuid.uuid4().hex[:8], account_id=a.id,
                                        action="disconnected", detail="사용자 해제"))
    s.commit()
    return ConnectorStatusOut.from_account(a, ConnectorRegistry.mode_for(a.domain))


# ── 로컬 우선: 수동 임포트(CSV / .ics / HealthKit export) ──
@router.post("/connectors/{account_id}/import", response_model=ImportResultOut)
async def import_file(account_id: str, file: UploadFile = File(...),
                      s: Session = Depends(get_session)):
    a = s.get(ConnectorAccount, account_id)
    if not a: raise HTTPException(404, "not found")
    content = await file.read()
    conn = ConnectorRegistry.get(s, a)              # csv/ics/healthkit 모드 커넥터
    if not hasattr(conn, "import_bytes"):
        raise HTTPException(400, "이 커넥터는 파일 임포트를 지원하지 않아요")
    res = conn.import_bytes(s, content, filename=file.filename)   # 내부에서 fetch→normalize→write
    return ImportResultOut(**res)
```

> **연결 흐름(real)**: 프론트가 `GET /api/connectors/oauth/start?domain=mail&provider=gmail` 으로 `authorize_url` 을 받아 새 창으로 이동 → 사용자 동의 → 제공자가 `GET /api/connectors/oauth/callback?code=&state=` 호출 → 토큰 교환·암호화 저장·초기 full sync → `frontend_origin/life?connect=ok` 로 리다이렉트. **로컬 우선 흐름(import)**: `POST /api/connectors/{id}/import` 멀티파트 업로드(CSV/.ics/HealthKit export) → OAuth 없이 즉시 정규화·저장.

---

### 3.10 능동 레이어 — `worker/sync_jobs.py` (주기 sync)

phase-7 의 `backend/app/worker/`(APScheduler)에 주기 동기화 잡을 등록한다. 프로토타입은 수동 트리거(`/api/connectors/sync-all`)도 제공(데모 결정성).

```python
# backend/app/worker/sync_jobs.py
from ..config import get_settings
from ..db import engine
from sqlmodel import Session
from ..connectors.registry import ConnectorRegistry
from ..models import ConnState


def run_periodic_sync():
    """worker 스케줄러가 sync_interval_minutes 마다 호출. real/연결된 계정만 sync."""
    with Session(engine) as s:
        for a in ConnectorRegistry.accounts(s):
            if a.state in (ConnState.connected, ConnState.error) and a.mode.value != "mock":
                ConnectorRegistry.get(s, a).sync(s)


def register(scheduler):
    """phase-7 worker.main 에서 호출."""
    mins = get_settings().sync_interval_minutes
    scheduler.add_job(run_periodic_sync, "interval", minutes=mins,
                      id="connector_sync", replace_existing=True)
```

> mock 계정은 주기 sync 대상이 아니다(데이터가 변하지 않음). real/연결 계정만 `sync_interval_minutes`(기본 15분) 간격으로 증분 sync → 새 객체에서만 이벤트 발행 → 자동화/알림 작동.

---

## 4. 데이터/타입/API 계약

### 4.1 신규 엔드포인트 (prefix `/api`)

| 메서드 | 경로 | 용도 |
|---|---|---|
| GET | `/api/connectors?domain=` | 커넥터 계정 상태 목록(`ConnectorStatusOut[]`) — 라이프 카드/설정에서 사용 |
| POST | `/api/connectors/{account_id}/sync?full=` | 단일 계정 수동 동기화 → `SyncResultOut` |
| POST | `/api/connectors/sync-all` | 연결된 전 계정 동기화 → `SyncResultOut[]` |
| GET | `/api/connectors/oauth/start?domain=&provider=&redirect_after=` | OAuth 인증 URL 발급 → `OAuthStartOut` |
| GET | `/api/connectors/oauth/callback?code=&state=` | OAuth 콜백(토큰 교환·저장·초기 sync) → 프론트 리다이렉트 |
| POST | `/api/connectors/{account_id}/disconnect` | 연결 해제(토큰 폐기) → `ConnectorStatusOut` |
| POST | `/api/connectors/{account_id}/import` (multipart) | CSV/.ics/HealthKit export 수동 임포트 → `ImportResultOut` |

### 4.2 Pydantic 스키마 (`schemas.py` 추가, `types.ts` 와 1:1)

```python
# backend/app/schemas.py  (추가)
from datetime import datetime
from typing import Optional
from pydantic import BaseModel

class ConnectorStatusOut(BaseModel):
    id: str
    domain: str
    provider: str
    name: str                     # "회사" / "우리카드" / "Apple Health"
    kind: str                     # "Lumi Workspace" / "신용/체크" / "건강 앱"
    tone: str                     # blue|violet|coral|green|amber|ink|faint
    mode: str                     # mock | csv | healthkit | real
    env_mode: str                 # CONNECTOR_<DOMAIN> 값(real 허용 여부 표시)
    state: str                    # connected|disconnected|syncing|error|token_expired
    on: bool                      # state==connected (라이프 카드 on 호환)
    last: str                     # last_label ("방금 동기화"/"오늘 09:12"/"연결 안 됨")
    last_synced_at: Optional[datetime] = None
    error_detail: str = ""

    @classmethod
    def from_account(cls, a, env_mode: str) -> "ConnectorStatusOut":
        return cls(id=a.id, domain=a.domain, provider=a.provider, name=a.name,
                   kind=a.kind, tone=a.tone, mode=a.mode, env_mode=env_mode,
                   state=a.state, on=(a.state == "connected"), last=a.last_label,
                   last_synced_at=a.last_synced_at, error_detail=a.error_detail)

class OAuthStartOut(BaseModel):
    authorize_url: str

class SyncResultOut(BaseModel):
    domain: str
    account_id: str
    seen: int = 0
    upserted: int = 0
    skipped: int = 0
    errors: int = 0
    events_published: list[str] = []
    detail: str = ""

class ImportResultOut(BaseModel):
    entity_type: str
    imported: int = 0
    skipped: int = 0
    errors: int = 0
    detail: str = ""
```

```ts
// frontend/lib/types.ts  (추가 — 백엔드 키 snake_case 유지)
export type ConnState =
  | "connected" | "disconnected" | "syncing" | "error" | "token_expired";

export interface ConnectorStatus {
  id: string;
  domain: "mail" | "calendar" | "chat" | "finance" | "health" | "knowledge";
  provider: string;
  name: string;
  kind: string;
  tone: "blue" | "violet" | "coral" | "green" | "amber" | "ink" | "faint";
  mode: "mock" | "csv" | "healthkit" | "real";
  env_mode: string;
  state: ConnState;
  on: boolean;
  last: string;
  last_synced_at: string | null;
  error_detail: string;
}
export interface OAuthStart { authorize_url: string; }
export interface SyncResult {
  domain: string; account_id: string;
  seen: number; upserted: number; skipped: number; errors: number;
  events_published: string[]; detail: string;
}
export interface ImportResult {
  entity_type: string; imported: number; skipped: number; errors: number; detail: string;
}
```

### 4.3 요청/응답 JSON 예시

`GET /api/connectors?domain=health` (시드 mock — `life-data.js: sources.health` 그대로):

```json
[
  {"id":"ca-health-apple","domain":"health","provider":"apple_health","name":"Apple Health",
   "kind":"건강 앱","tone":"coral","mode":"mock","env_mode":"mock","state":"connected",
   "on":true,"last":"방금 동기화","last_synced_at":null,"error_detail":""},
  {"id":"ca-health-watch","domain":"health","provider":"apple_watch","name":"Apple Watch",
   "kind":"웨어러블","tone":"violet","mode":"mock","env_mode":"mock","state":"connected",
   "on":true,"last":"1분 전","last_synced_at":null,"error_detail":""},
  {"id":"ca-health-fit","domain":"health","provider":"google_fit","name":"Google Fit",
   "kind":"활동 추적","tone":"green","mode":"mock","env_mode":"mock","state":"disconnected",
   "on":false,"last":"연결 안 됨","last_synced_at":null,"error_detail":""}
]
```

`POST /api/connectors/ca-mail-personal/sync` (Gmail real, 새 메일 2건):

```json
{
  "domain": "mail", "account_id": "ca-mail-personal",
  "seen": 12, "upserted": 2, "skipped": 10, "errors": 0,
  "events_published": ["mail.received", "mail.received"],
  "detail": ""
}
```

`GET /api/connectors/oauth/start?domain=calendar&provider=google_calendar`:

```json
{ "authorize_url": "https://accounts.google.com/o/oauth2/v2/auth?client_id=...&scope=https%3A%2F%2Fwww.googleapis.com%2Fauth%2Fcalendar.events&state=...&code_challenge=...&code_challenge_method=S256" }
```

`POST /api/connectors/ca-fin-woori/import` (우리카드 CSV 업로드, multipart):

```json
{ "entity_type": "finance_tx", "imported": 47, "skipped": 3, "errors": 0,
  "detail": "우리카드 6월 명세 · 식비 412,000 · 쇼핑 298,000 정규화 완료" }
```

---

## 5. 디자인 충실도 노트

기준 파일(REF): `design-reference/assets/life-data.js`, `mail-data.js`, `cal-data.js`, `life.css`, `dash.css`(토큰).

### 5.1 라이프 connector 카드 — `sources` 의 on/last 그대로

원본 `life-data.js: sources` 의 각 항목은 `{id, name, kind, tone, on, last}` 다. 이 carded UI 를 Phase 13 이 **실제 커넥터 상태로** 채운다(값은 mock 일 때 원본과 1바이트 동일).

```
원본 life-data.js:
  { id:"apple",  name:"Apple Health", kind:"건강 앱",  tone:"coral",  on:true,  last:"방금 동기화" }
  { id:"toss",   name:"토스",          kind:"계좌 통합", tone:"coral",  on:true,  last:"오늘 08:40" }
  { id:"fit",    name:"Google Fit",   kind:"활동 추적", tone:"green",  on:false, last:"연결 안 됨" }
→ ConnectorStatusOut: name/kind/tone 동일, on=state=="connected", last=last_label
```

`ConnectorCard.tsx` 는 phase-11 의 라이프 카드를 재사용한다. 톤은 **키**(`"coral"` 등)로 받아 `var(--coral)`(=`#df7256`, dash.css) 로 변환(CONTRACT: tone 은 백엔드 키 저장, 프론트 `var(--tone)` 변환). `on:false` 카드는 흐리게(`--faint #b6b3ad`) + "연결 안 됨" + [연결] 버튼.

### 5.2 메일 계정 톤·라벨 보존

`mail-data.js: accounts` 의 3계정 톤(`work=coral`, `personal=blue`, `side=violet`)·`kind`(Lumi Workspace/Google/HEY)·`unread` 배지를 그대로 유지. 정규화 라벨은 원본 메일 `labels`(`["디자인"]`, `["경영진"]`, `["금융"]`, `["채용"]`, `["영수증"]` …)와 동일 한국어 라벨로 매핑.

### 5.3 상태 배지 색 (dash.css 액센트 그대로)

| 상태 | 라벨 | tone/색 |
|---|---|---|
| connected | "방금 동기화" / "오늘 09:12" / "1분 전" | `--green #4e9b66` 점 |
| syncing | "동기화 중…" | `--blue #4f72e0` (스피너) |
| disconnected | "연결 안 됨" | `--faint #b6b3ad` |
| error | "동기화 실패 · 다시 시도" | `--coral #df7256` |
| token_expired | "권한 만료 · 다시 연결" | `--amber #e0a23c` |

> 글래스 카드는 `background: var(--glass)` + `backdrop-filter: var(--blur)`(blur(26px) saturate(190%)) + `border: 1px solid var(--glass-brd)`(dash.css §11.4). 라운드 `--radius 22px`. 라이트/다크는 `data-theme`(next-themes) 그대로.

### 5.4 캘린더/건강/금융 표시 포맷 보존

- 일정: `cal-data.js` 의 `start/end`("09:30"), `loc`("Zoom"/"대회의실"), `soon:true`(분기 전략 미팅) 보존. real 동기화 시에도 동일 키.
- 건강: "7,240보 / 10,000보 / 72%", "수면 7h 12m", "안정 심박 62 bpm" 등 `.mono`(tabular-nums) 표시.
- 금융: 금액 정수 원화(`1,284,000`), 카테고리 6종, 구독 spark — `life-data.js: finance` 포맷 그대로.

---

## 6. 상태 처리 & 엣지 케이스

| 상황 | 기대 동작 | 구현 포인트 |
|---|---|---|
| **로딩** | 커넥터 목록/sync 중 스켈레톤 + "동기화 중…" 배지 | `state=syncing`, 프론트 낙관적 표시 |
| **빈 상태** | 도메인에 연결 계정 0개 | "아직 연결된 OO이 없어요 · [연결하기]" 빈-상태 카드 |
| **연결(connect)** | OAuth 동의 후 `connect=ok` 리다이렉트 + 초기 full sync | `oauth_callback` 에서 1회 full sync |
| **해제(disconnect)** | 토큰 폐기(`token_enc=""`), state→disconnected, last→"연결 안 됨" | `/disconnect`, 데이터는 보존(이미 받은 메일 유지) |
| **오류(error)** | sync 예외 → state=error, error_detail 저장, **서버 안 죽음** | `BaseConnector.sync` 의 `except` rollback + `_finish_error` |
| **동기화 중(syncing)** | 재진입 sync 방지 권장(같은 계정 동시 sync) | `state==syncing` 이면 409 또는 skip |
| **권한 만료(token_expired)** | refresh 실패 → state=token_expired, "다시 연결" CTA | `valid_access_token` 의 PermissionError → sync error → UI 재인증 |
| **레이트리밋(429)** | Retry-After 만큼 백오프 후 재시도, 초과 시 부분 결과 commit | `RateLimiter.handle_response` |
| **중복/충돌** | 같은 external_id 재수신 → upsert(중복 0). etag 동일 → skip | `external_link` UNIQUE + `_is_duplicate` |
| **오프라인(Ollama/네트워크 끊김)** | mock 은 정상, real 은 fetch 실패→error 로 표시(데모는 mock 폴백) | env 기본 mock, real 실패가 앱을 멈추지 않음 |
| **CSV 형식 오류** | 파싱 실패 행 skip + `errors` 카운트, 정상 행만 import | `csv_import.import_bytes` 행 단위 try/except |
| **암호화 키 손상/회전** | `decrypt_token` → `{}` → token_expired (재연결 유도) | `crypto.decrypt_token` InvalidToken 폴백 |
| **mock↔real 공존** | 메일=real, 금융=mock 동시 가능 | 계정별 `mode`, env 는 도메인 게이트 |
| **시드 mock 보호** | env=real 이어도 시드 mock 계정은 mock 유지(데모 결정성) | `ConnectorRegistry.get` 모드 결정 정본 |

엣지 메모:
- `GOOGLE_CLIENT_ID` 미설정인데 `/oauth/start` 호출 → 400 "real 연동 비활성"(앱 안 죽음).
- `.ics`/CSV 임포트는 OAuth 불필요 → 클라이언트 비밀 없이도 **로컬 우선** 데이터 유입 보장.
- 재동기화로 같은 메일이 들어와도 `created=False` → `mail.received` **재발행 안 함**(중복 알림 방지).

---

## 7. 연합 이벤트 (발행/구독)

CONTRACT 의 `event_bus`(phase-7 정본) 이벤트 모델을 사용한다. **이 phase 의 핵심 가치**: mock 에서는 시뮬레이션이던 이벤트가 real 에서 **실제 외부 데이터로 발행**되어, phase 7~12 의 구독자가 진짜 데이터로 동작한다.

### 7.1 이 phase가 발행하는 이벤트

| 이벤트 | 발행 시점 | 페이로드 | 주 구독자 |
|---|---|---|---|
| `mail.received` | mail sync 가 새 email upsert(`created=True`) | `{account_id, entity_type:"email", entity_id}` | phase-9(메일 분석 → `email.ai.tasks/events`), 자동화 `evaluator`(영수증/뉴스레터 정리 규칙), 알림 트리아지 |
| `calendar.updated` | calendar sync 가 새/변경 event upsert | `{account_id, entity_type:"event", entity_id}` | phase-8(일정 표시·회의 도우미), 집중 모드 빈 블록 재계산, 리스크 레이더 |
| `finance.transaction` | finance sync/CSV import 가 새 거래 | `{account_id, entity_type:"finance_tx", entity_id}` | phase-11(예산/카테고리 갱신), 자동화(예산 초과·잠자는 구독 제안 → 결재함) |
| `health.sample` | health sync/HealthKit import 가 새 샘플 | `{account_id, entity_type:"health_sample", entity_id}` | phase-11(rings/sleep 갱신, 코치 리마인더) |
| `knowledge.ingested` | knowledge sync 가 새 knowledge_item | `{account_id, entity_type:"knowledge_item", entity_id}` | phase-10 RAG ingest(chunk→embed), 리서치 Q&A·라이프 지식 |
| `chat.message` | chat webhook 수신 | `{account_id, entity_type:"chat_message", entity_id}` | phase-9 알림 트리아지 |
| `connector.synced` | 모든 도메인 sync 완료 | `{account_id, upserted, skipped, errors}` | phase-12 하루 마감(동기화 요약), 라이프 카드 last 갱신 |
| `connector.error` | sync 실패/토큰 만료 | `{account_id, detail, state}` | 알림(재연결 안내), 라이프 카드 error 배지 |

### 7.2 이 phase가 구독하는 이벤트

| 구독 이벤트 | 출처 | 처리 |
|---|---|---|
| `automation.matched` (mail/calendar 동작) | phase-7 evaluator | `BaseConnector.write()`(예: calendar_write tool → real 캘린더에 이벤트 쓰기) — 양방향(write 메서드) |
| `worker.tick` (스케줄러) | phase-7 worker | `run_periodic_sync()` 실행(증분 sync) |

### 7.3 연합 시나리오 (PROJECT-README §5 — 실데이터 버전)

```
[real Gmail 연결됨] 현우님 "온보딩 시안 v3" 메일 도착
   → GmailConnector.sync(): fetch → normalize_email → write(email) created=True
   → event_bus.publish("mail.received", {entity_id:"m-xxxx"})
   → phase-9 메일 분석(로컬 LLM): ai.tasks=[{온보딩 시안 v3 피드백 정리, due 6/13, 높음}]
       ai.events=[], ai.replies=[{수락·일정 약속, ...}]   (mail-data.js m1 의 ai 묶음과 동일 형태)
   → 자동화 evaluator: "검토 요청 메일 → 작업 초안" 규칙 매칭
   → autonomy_setting=mixed, risk=low → 작업 자동 생성(되돌리기) → 결재함 a 항목
   → 작업 페이지 GET /api/tasks 에 등장 (federation, 개인=프로젝트 원칙)
   → 하루 마감: "아리가 처리" 카운트 +1, 아낀 시간 합산
```

> 이 흐름은 mock(`MockMailConnector.simulate_new()`)에서도 **동일 코드**로 재현된다 — 그래서 데모/테스트가 결정적이다. real 은 같은 이벤트를 진짜 메일에서 만들 뿐이다(인터페이스 불변의 증거).

---

## 8. 테스팅 & 검증

### 8.1 실행 명령

```bash
# 의존성 추가(uv)
cd /Users/kim1634/mycloud/personal/workspace/ari_assistant/workspace/backend
uv add cryptography
uv add --dev respx
# (pip 폴백: pip install cryptography && pip install respx)

# 마이그레이션 + 시드(커넥터 mock 계정 포함)
uv run alembic upgrade head
uv run python -m app.seed         # run_seed() → _seed_connectors() 포함

# 백엔드 테스트
uv run pytest tests/test_connector_contract.py tests/test_connector_swap.py \
              tests/test_oauth_mock.py tests/test_sync_idempotent.py \
              tests/test_normalize_golden.py tests/test_crypto.py \
              tests/test_federation_real.py -q
uv run pytest -q                  # 전체(기존 phase 회귀 포함)

# 수동 스모크
curl -s http://localhost:8000/api/connectors?domain=health | python3 -m json.tool
curl -s -X POST http://localhost:8000/api/connectors/ca-mail-work/sync | python3 -m json.tool

# 프론트
cd ../frontend && pnpm test
pnpm exec playwright test playwright/connectors.spec.ts
```

### 8.2 테스트 케이스 목록

| # | 케이스 | 방법 | 통과 기준 |
|---|---|---|---|
| C1 | **인터페이스 계약**: 모든 도메인 mock/real 이 `fetch/normalize/write/sync` 구현 | `test_connector_contract.py` — 각 클래스가 `BaseConnector` 서브클래스 + 메서드 존재 | 6개 도메인 × (mock + real/alt) 전부 통과, 추상 미구현 0 |
| C2 | **Mock↔Real 스왑 동일 shape**: `CONNECTOR_MAIL=mock` vs `real`(fake) 에서 `/api/connectors` 응답 키 동일 | `test_connector_swap.py` — env monkeypatch + 응답 스키마 비교 | `ConnectorStatusOut` 키/타입 동일, 페이지 코드 변경 0 |
| C3 | **OAuth 모킹**: start→callback→token 저장(암호화) | `test_oauth_mock.py` — `respx` 로 token 엔드포인트 모킹 | `connector_account.token_enc` 비어있지 않고 평문 access_token **미포함**, state row 소거 |
| C4 | **동기화 멱등성**: 같은 페이로드 2회 sync → 중복 0 | `test_sync_idempotent.py` — fake fetch 동일 레코드 2회 | 1회차 upserted=N, 2회차 upserted=0 & skipped=N, `external_link` row N개 |
| C5 | **정규화 골든**: Gmail 픽스처 → 고정 `email.fields` | `test_normalize_golden.py` — 픽스처 vs 기대 dict | 필드별 정확 매핑(subject/from/labels/read/starred) |
| C6 | **토큰 암복호화 왕복** | `test_crypto.py` | `decrypt(encrypt(x))==x`, 잘못된 키 → `{}` (예외 누수 없음) |
| C7 | **연합(real fake) → 이벤트 발행** | `test_federation_real.py` — fake GmailConnector sync → event_bus 스파이 | 새 메일당 `mail.received` 1회, 재sync 시 0회 |
| C8 | **토큰 만료 → token_expired** | refresh 401 모킹 | state=token_expired, sync errors≥1, 서버 200(앱 안 죽음) |
| C9 | **레이트리밋 백오프**: 429 → Retry-After 후 재시도 | `RateLimiter` 단위 + 429 모킹 | 백오프 호출 발생, 최종 성공 |
| C10 | **CSV 임포트(로컬 우선)**: 우리카드 CSV → finance_tx | `import_bytes` 단위 + 오류 행 skip | imported>0, 오류 행 skip+errors 카운트 |
| C11 | **.ics 임포트**: 이벤트 → `event` 정규화 | `IcsConnector.import_bytes` | `cal-data.js` 키로 귀결 |
| C12 | **disconnect 토큰 폐기** | `/disconnect` 호출 | `token_enc==""`, state=disconnected, 기존 데이터 보존 |
| C13 | **mock 시드 보호**: env=real 인데 시드 mock 계정 | registry 모드 결정 | 시드 계정 mock 유지(데모 결정성) |
| C14 | **회귀**: phase 2~12 기존 테스트 | `uv run pytest -q` | 전부 green(인터페이스 불변 증명) |

### 8.3 대표 테스트 코드

```python
# backend/tests/test_sync_idempotent.py
from app.connectors.mail.mock import MockMailConnector
from app.models import ConnectorAccount, ExternalLink, Email
from sqlmodel import select

def test_sync_is_idempotent(session, seed):
    acct = session.get(ConnectorAccount, "ca-mail-work")
    conn = MockMailConnector(acct).simulate_new()       # 새 메일 1건 흘림
    r1 = conn.sync(session)
    conn2 = MockMailConnector(acct).simulate_new()      # 같은 external_id 재공급
    r2 = conn2.sync(session)
    assert r1.upserted == 1 and r2.upserted == 0 and r2.skipped == 1
    links = session.exec(select(ExternalLink).where(
        ExternalLink.account_id == "ca-mail-work")).all()
    assert len([l for l in links if l.external_id == "sim-m1"]) == 1   # 중복 0
```

```python
# backend/tests/test_oauth_mock.py
import respx, httpx
from app.connectors import oauth
from app.crypto import decrypt_token

@respx.mock
def test_oauth_callback_stores_encrypted_token(session, monkeypatch):
    monkeypatch.setenv("GOOGLE_CLIENT_ID", "cid"); monkeypatch.setenv("GOOGLE_CLIENT_SECRET", "sec")
    url = oauth.start_oauth(session, "mail", "gmail")
    state = url.split("state=")[1].split("&")[0]
    respx.post("https://oauth2.googleapis.com/token").mock(
        return_value=httpx.Response(200, json={
            "access_token": "AT", "refresh_token": "RT", "expires_in": 3600, "scope": "...readonly"}))
    acct = oauth.finish_oauth(session, code="CODE", state=state)
    assert acct.mode == "real" and acct.state == "connected"
    assert "AT" not in acct.token_enc            # 평문 노출 금지
    assert decrypt_token(acct.token_enc)["access_token"] == "AT"
```

```python
# backend/tests/test_connector_swap.py
def test_status_shape_same_mock_and_real(client, monkeypatch):
    monkeypatch.setenv("CONNECTOR_MAIL", "mock")
    mock_keys = set(client.get("/api/connectors?domain=mail").json()[0].keys())
    monkeypatch.setenv("CONNECTOR_MAIL", "real")
    real_keys = set(client.get("/api/connectors?domain=mail").json()[0].keys())
    assert mock_keys == real_keys     # 페이지가 보는 shape 불변
```

### 8.4 수동 QA 체크리스트

- [ ] `python -m app.seed` 후 `/api/connectors` 가 mail 3 + calendar 1 + health 3 + finance 4 + knowledge 3 = **14 계정** 반환, 값이 `mail-data.js`/`life-data.js` 와 동일.
- [ ] Google Fit·KB증권은 `on:false` + `last:"연결 안 됨"`.
- [ ] 라이프 페이지 connector 카드가 `/api/connectors` 값으로 렌더(톤·last 문구 원본 일치).
- [ ] `CONNECTOR_*` 전부 mock(기본)에서 모든 페이지가 phase 7~12 와 **동일하게** 동작(회귀 없음).
- [ ] `GOOGLE_CLIENT_ID` 미설정 시 `/oauth/start?provider=gmail` → 400(앱 안 죽음), mock 페이지 정상.
- [ ] (자격증명 있을 때) OAuth start → 동의 → callback → `connect=ok` 리다이렉트 + 초기 sync → 메일 등장.
- [ ] 같은 계정 sync 2회 → 두 번째 `upserted:0`(중복 없음).
- [ ] CSV(우리카드 명세) 임포트 → 금융 거래/카테고리 갱신, OAuth 불필요.
- [ ] `.ics` 임포트 → 일정에 이벤트 등장, OAuth 불필요.
- [ ] sync 실패(네트워크 끊김) → 카드 error 배지 + "다시 시도", **서버 200·앱 살아있음**.
- [ ] 토큰 만료 시 "권한 만료 · 다시 연결"(amber) 배지.
- [ ] DB 의 `connector_account.token_enc` 에 **평문 토큰이 보이지 않음**(grep 으로 access_token 미검출).
- [ ] disconnect 후 토큰 폐기되지만 이미 받은 메일/일정은 유지.

### 8.5 통과 기준 (요약)

C1~C14 전부 green + 수동 체크리스트 전 항목. **핵심 합격선**: (1) `CONNECTOR_*=mock` 기본에서 phase 7~12 회귀 0(인터페이스 불변), (2) real(fake) sync 가 동일 코드 경로로 `mail.received`/`calendar.updated` 를 발행, (3) 동기화 멱등(중복 0), (4) 토큰 평문 미저장, (5) 모든 외부 실패가 앱을 멈추지 않음(mock/로컬 폴백).

---

## 9. 완료 기준 (Definition of Done)

- [ ] `backend/app/connectors/`: `base.py`(fetch/sync/normalize/write 추상 + 공통 upsert/멱등/이벤트), `registry.py`(CONNECTOR_<DOMAIN> 선택), `oauth.py`(Auth Code+PKCE start/callback/refresh), `ratelimit.py`, `crypto.py` 존재.
- [ ] 6개 도메인(mail/calendar/chat/finance/health/knowledge) 각각 `mock.py` + `real_*.py`(또는 csv/ics/healthkit alt) + `normalize.py` 존재, 모두 `BaseConnector` 계약 충족.
- [ ] 신규 5개 테이블(`connector_account`, `connector_sync_state`, `external_link`(복합 UNIQUE), `oauth_state`, `connector_account_log`) Alembic 마이그레이션 + 왕복 OK.
- [ ] `_seed_connectors()` 가 `run_seed()` 에 통합되어 14개 mock 계정 시드(값 = `mail-data.js`/`life-data.js`/`cal-data.js` 원본).
- [ ] `config.py` 에 `CONNECTOR_<DOMAIN>`(기본 mock) + OAuth/암호화 env, `.env` 예시 갱신.
- [ ] 라우터 `/api/connectors`(상태/sync/sync-all/oauth start·callback/disconnect/import) + `main.py` `include_router(prefix="/api")`.
- [ ] **mock 기본에서 phase 7~12 전부 회귀 0**(코드 변경 없이 동작) — 인터페이스 불변 증명.
- [ ] real(fake) sync → `event_bus` 에 `mail.received`/`calendar.updated`/… 발행 → 자동화/작업/알림이 실데이터로 동작.
- [ ] 증분 sync(cursor) + 정규화(provider→내부) + 멱등 upsert(external_link UNIQUE) + 레이트리밋(429 백오프) 동작.
- [ ] 로컬 우선: CSV/.ics/HealthKit export 수동 임포트가 OAuth 없이 동작.
- [ ] 토큰 암호화 저장(평문 미저장) + 만료 자동 refresh + 실패 시 token_expired.
- [ ] 라이프 connector 카드 on/last/상태(connected/disconnected/syncing/error/token_expired) 디자인 충실 렌더(tone=키→`var(--tone)`).
- [ ] C1~C14 테스트 green + 수동 QA 체크리스트 전 항목.

---

## 10. 다음 단계

- **다음 문서: `phase-14-proactive-agent.md`** — 능동형 에이전트(심부름 대행) + 멀티모달 캡처(음성 STT/이미지 Vision 실구현) + 능동 알림. Phase 13 의 RealConnector 가 공급하는 **실데이터 이벤트 스트림**(`mail.received`/`calendar.updated`/`finance.transaction`/`health.sample`) 위에서, `backend/app/agents/`(plan→act→observe→reflect) 와 `backend/app/worker/`(다이제스트/브리핑/패턴 탐지)가 동작한다. 에이전트 tool(`calendar_write`/`task_create`/`form_fill`)은 본 phase 의 `BaseConnector.write()` 를 호출해 외부에 **쓰기**까지 연합한다.
- 이후 `phase-15-production.md` — 인증/멀티유저(per-user `connector_account.user_id` 스코프), 토큰 시크릿 매니저·키 회전, 배포/관측성(sync 지표·이벤트 추적)/성능/보안.

> 빌드 순서(CONTRACT): 7 → 8 → 9 → 10 → 11 → 12 → **13** → 14 → 15. 페이지는 mock-first 로 만들고, **13(이 문서)에서 real 로 전환**한다.