diff --git a/apps/server/src/shared-mailbox-runtime.ts b/apps/server/src/shared-mailbox-runtime.ts
index 5f15bf9..eac7c04 100644
--- a/apps/server/src/shared-mailbox-runtime.ts
+++ b/apps/server/src/shared-mailbox-runtime.ts
@@ -70,7 +70,7 @@ export class SharedMailboxRuntime implements SharedMailboxAdminPort {
     const source = this.opts.directorySources?.[provider];
     if (!source) throw new CommandError("unavailable", `directory provider가 구성되지 않았습니다: ${provider}`);
     const snapshot = await source.read(tenantId);
-    const input = await directorySyncInputFromSnapshot(this.opts.db, { tenantId, provider, now: Date.now(), snapshot });
+    const input = directorySyncInputFromSnapshot({ tenantId, provider, now: Date.now(), snapshot });
     await syncDirectorySnapshot(this.opts.db, input);
     return { data: { provider, identities: snapshot.identities.length, groups: snapshot.groups.length }, message: "directory snapshot 동기화 완료" };
   }
diff --git a/docs/OPERATIONS.md b/docs/OPERATIONS.md
index 49b3de6..498184c 100644
--- a/docs/OPERATIONS.md
+++ b/docs/OPERATIONS.md
@@ -368,6 +368,19 @@ for n in mx.ionosphere.test mta-sts.ionosphere.test admin.ionosphere.test; do
 done
 ```
 
+### 디렉터리 외부 키 재연결 방지
+
+LDAP/AD `directory-identity-link`도 OIDC와 같은 최초 주인 규칙을 쓴다. 외부 키를 처음 연결하면
+`identity_claims`에 테넌트·공급자·외부 키 해시·계정 id를 남긴다. 연결을 끊거나 계정을 삭제해도 묘비는
+지우지 않으므로, 같은 외부 키는 원래 계정에만 다시 연결할 수 있다. 계정 삭제 뒤에는 원래 계정도 없으므로
+새 계정에 재사용할 수 없다. 외부 키 원문(DN 등)은 영구 기록에 남기지 않는다.
+
+이 기능이 생기기 전에 이미 연결된 행도 안전하다. `directory-identity-unlink`와 계정 드레인이 연결 행을
+지우는 같은 DB 배치에서 먼저 최초 주인을 기록한다.
+
+외부 키 해시는 원문을 직접 남기지 않는 **가명화**다. LDAP DN처럼 후보가 좁은 값은 DB를 읽을 수 있는 사람이
+후보를 해시해 대조할 수 있다. 서버 비밀 HMAC은 키 회전 때 묘비를 무효화하므로 쓰지 않는다.
+
 ### 외부 OIDC 로그인 (E3)
 
 관리 콘솔과 셀프서비스에 "IdP로 로그인" 버튼을 얹는다. 기본으로 꺼져 있다.
diff --git a/docs/PROTOCOLS.md b/docs/PROTOCOLS.md
index 71c6e9c..00875e6 100644
--- a/docs/PROTOCOLS.md
+++ b/docs/PROTOCOLS.md
@@ -352,7 +352,7 @@ SPF+DKIM pass · From 도메인 정렬(SPF 또는 DKIM) · DMARC p=none 이상 
 | 표면 제한 | `credentials.scopes` — 인증 함수 안의 단일 관문, 모르는 이름은 fail closed | `credentialAllowsSurface` |
 | 시도 제한 | 실패만 셈, IP 프리픽스·계정 두 축(모든 메일 프로토콜·웹), 5회 뒤 1초부터 두 배·상한 15분·15분 뒤 초기화(공통 수치). 성공하면 계정 축만 지움. 서버들이 공유 DB로 함께 센다(DB 장애 시 프로세스 카운터) | `core/auth-throttle.ts`, `store/auth-failures.ts` |
 | 정지 | `account-suspend` — 새 인증·제출·수신 차단, 데이터 보존, 되돌릴 수 있음 | — |
-| 삭제 | `account-delete` — 즉시 비노출, 리퍼가 24시간 안에 메일·자격증명·sieve를 지우고 계정 행까지 없앤다(`ionosphere_account_drain_overdue`). 끝날 때까지 주소 재등록 차단 | SCHEMA §7-7 |
+| 삭제 | `account-delete` — 즉시 비노출, 리퍼가 24시간 안에 메일·자격증명·sieve를 지우고 계정 행까지 없앤다(`ionosphere_account_drain_overdue`). 끝날 때까지 주소 재등록 차단. SSO sub·LDAP/AD 외부 키 해시는 최초 주인 묘비로 남긴다 | SCHEMA §7-7 |
 | 반영 지연 | 새 인증은 즉시. **JMAP은 인증 캐시로 최대 30초**(외부 토큰은 exp를 넘지 않는다). 제출(SMTP)은 메시지마다 상태를 다시 본다. 열린 IMAP·POP3·ManageSieve 연결은 **다음 명령에서** 계정 상태를 다시 본다(G2). 외부 토큰으로 연 연결은 토큰 exp에 닫힌다(E4) | `jmap-server.ts`, `backend.ts`, `proto-*/server.ts`, `core/auth-expiry.ts` |
 | 디렉터리 로그인 | 기본 꺼짐. 로컬 실패 뒤 폴백, 관리자가 연결한 불변 키만. provider별 허용 표면(`surfaces`, 생략하면 전부) — 밖이면 bind 전에 거절 | `shared-mailbox-runtime.ts`, `directory-config.ts` |
 | 관리 주체 | 사람: `admin_users`(메일 계정과 별개, TOTP 필수, 서버 세션). 서비스: `api_keys`(read/write/admin). 정적 `rootToken`은 부트스트랩 전용(활성 owner가 생기면 경고) | `store/admin-auth.ts`, `api/server.ts` |
diff --git a/docs/SCHEMA.md b/docs/SCHEMA.md
index ca5a4a4..39d99ae 100644
--- a/docs/SCHEMA.md
+++ b/docs/SCHEMA.md
@@ -403,9 +403,9 @@ CREATE INDEX ix_auth_failures_last ON auth_failures(last_fail_at);
 
 ```sql
 CREATE TABLE identity_claims (
-  realm        VARCHAR(16) NOT NULL,   -- 'admin'(admin_identities) | 'account'(directory_identities의 oidc: 행)
-  provider     VARCHAR(64) NOT NULL,   -- 설정의 공급자 id(issuer를 고정한다). account 영역은 'oidc:<id>'
-  subject      VARCHAR(255) NOT NULL,  -- IdP sub
+  realm        VARCHAR(16) NOT NULL,   -- 'admin' | 'account'(OIDC) | 'directory'(LDAP/AD)
+  provider     VARCHAR(64) NOT NULL,   -- OIDC 공급자 id / directory는 '<tenant id>:<provider>'
+  subject      VARCHAR(255) NOT NULL,  -- IdP sub / directory는 외부 키의 'sha256:<hex>'
   owner_id     VARCHAR(26) NOT NULL,   -- 처음 연결한 주체(관리자 또는 메일 계정 id)
   claimed_at   BIGINT NOT NULL,
   PRIMARY KEY (realm, provider, subject)
@@ -420,8 +420,11 @@ sub를 **처음 연결할 때** 주인을 기록하고 지우지 않는다. 연
 - 결정의 "iss+sub"는 공급자 id+sub로 대신한다. 같은 영역(account면 같은 테넌트)에서 같은 issuer를 두 id로 쓰는 설정은
   기동이 거부한다(`oidc-config.ts`). ⚠ 공급자 id를 바꾸거나 다른 IdP에 다시 쓰지 말 것.
 - account 영역도 테넌트와 무관하게 전역이다. 개인정보가 아니라 불투명 식별자뿐이라 탈퇴 뒤에도 남긴다.
-- LDAP/AD 디렉터리 연결(`directory-identity-*`)은 대상이 아니다 — SSO 식별자가 아니고, objectGUID는 IdP가 재활용하지 않는다.
-- 029는 이미 있는 연결을 그 주인으로 옮겨 적는다(NOT EXISTS로 멱등).
+- LDAP/AD 디렉터리 연결도 realm=`directory`로 같은 표를 쓴다(2026-10-19 결정). 계정 드레인이나 연결 해제 뒤
+  다음 동기화가 같은 외부 키를 만들더라도 다른 계정에 잇지 못한다. 범위는 테넌트+공급자이며, DN 같은 개인정보가
+  영구 묘비에 원문을 직접 남기지 않게 외부 키 대신 SHA-256을 저장한다. 이는 가명화라 후보 DN을 아는 사람의 대조까지
+  막지는 않는다. 이 기능 전의 연결은 해제·드레인할 때 먼저 claim한다.
+- 029는 당시 있던 OIDC 연결을 그 주인으로 옮겨 적는다(NOT EXISTS로 멱등).
 
 ## 5. DDL — 메일 스토어 코어
 
diff --git a/packages/admin-cmd/src/shared-mailbox.ts b/packages/admin-cmd/src/shared-mailbox.ts
index f390248..f271cc6 100644
--- a/packages/admin-cmd/src/shared-mailbox.ts
+++ b/packages/admin-cmd/src/shared-mailbox.ts
@@ -1,7 +1,5 @@
 import { CommandError, type Command, type CommandContext } from "./types.ts";
-import { ulid } from "@ionosphere/core";
-import { BatchConflictError } from "@ionosphere/db";
-import { directoryMembershipSource } from "@ionosphere/store";
+import { linkDirectoryIdentityWithClaim, unlinkDirectoryIdentityWithClaim } from "@ionosphere/store";
 
 function tenantId(ctx: CommandContext): string {
   if (!ctx.tenantId) throw new CommandError("invalid", "tenantId가 필요합니다");
@@ -51,20 +49,18 @@ export const sharedMailboxCommands: readonly Command[] = [
       if (otherLink.rows.length > 0) throw new CommandError("conflict", "이 account는 같은 provider의 다른 identity에 연결되어 있습니다");
       const linked = identity.rows[0]?.account_id;
       if (linked != null && String(linked) !== args.accountId) throw new CommandError("conflict", "directory identity가 다른 account에 연결되어 있습니다");
-      const now = Date.now();
-      try {
-        await ctx.db.batch([
-          { sql: "UPDATE directory_identities SET account_id = ? WHERE tenant_id = ? AND provider = ? AND external_key = ?", params: [args.accountId, scope, args.provider, args.externalKey] },
-          { sql: ctx.db.insertIgnore("principals", ["id", "tenant_id", "kind", "account_id", "provider", "external_key", "display_name", "created_at"]), params: [ulid(), scope, 0, args.accountId, args.provider, args.externalKey, identity.rows[0]?.display_name ?? null, now] },
-          { sql: "UPDATE principals SET account_id = ?, display_name = ? WHERE tenant_id = ? AND kind = 0 AND provider = ? AND external_key = ?", params: [args.accountId, identity.rows[0]?.display_name ?? null, scope, args.provider, args.externalKey] },
-          { sql: "UPDATE accounts SET permissions_version = permissions_version + 1 WHERE id = ? AND tenant_id = ?", params: [args.accountId, scope] },
-        ]);
-      } catch (error) {
-        // 사전 조회는 안내용일 뿐 동시 요청의 잠금이 아니다. 최종 승인은 DB UNIQUE 제약이 한다.
-        if (error instanceof BatchConflictError) {
-          throw new CommandError("conflict", "이 account는 같은 provider의 다른 identity에 연결되어 있습니다");
-        }
-        throw error;
+      const result = await linkDirectoryIdentityWithClaim(ctx.db, {
+        tenantId: scope,
+        provider: args.provider!,
+        externalKey: args.externalKey!,
+        accountId: args.accountId!,
+        displayName: identity.rows[0]?.display_name == null ? null : String(identity.rows[0].display_name),
+      });
+      if (result === "retired") {
+        throw new CommandError("conflict", "이 directory identity는 예전에 다른 account에 연결된 적이 있어 다시 이을 수 없습니다");
+      }
+      if (result === "conflict") {
+        throw new CommandError("conflict", "이 account는 같은 provider의 다른 identity에 연결되어 있거나, identity가 다른 account에 연결되어 있습니다");
       }
       return { data: { provider: args.provider, externalKey: args.externalKey, accountId: args.accountId }, message: "directory identity 연결 완료 — group membership 반영을 위해 다시 동기화하세요" };
     },
@@ -75,15 +71,12 @@ export const sharedMailboxCommands: readonly Command[] = [
       // OIDC 연결 행(`oidc:` 접두)은 OIDC 명령만 다룬다 — 여기서 바꾸면 그 명령의 공급자·충돌 검사를 우회한다(2026-10-18 검토).
       if ((args.provider ?? "").startsWith("oidc:")) throw new CommandError("invalid", "oidc: 공급자는 oidc-identity-link/unlink로 다루십시오");
       const scope = tenantId(ctx);
-      const identity = await ctx.db.query({ sql: "SELECT account_id FROM directory_identities WHERE tenant_id = ? AND provider = ? AND external_key = ?", params: [scope, args.provider, args.externalKey] });
-      const accountId = identity.rows[0]?.account_id == null ? null : String(identity.rows[0]?.account_id);
-      if (!accountId) throw new CommandError("notFound", "연결된 directory identity를 찾을 수 없습니다");
-      await ctx.db.batch([
-        { sql: "DELETE FROM account_memberships WHERE account_id = ? AND source = ?", params: [accountId, directoryMembershipSource(args.provider!)] },
-        { sql: "UPDATE principals SET account_id = NULL WHERE tenant_id = ? AND kind = 0 AND provider = ? AND external_key = ?", params: [scope, args.provider, args.externalKey] },
-        { sql: "UPDATE directory_identities SET account_id = NULL WHERE tenant_id = ? AND provider = ? AND external_key = ?", params: [scope, args.provider, args.externalKey] },
-        { sql: "UPDATE accounts SET permissions_version = permissions_version + 1 WHERE id = ? AND tenant_id = ?", params: [accountId, scope] },
-      ]);
+      const result = await unlinkDirectoryIdentityWithClaim(ctx.db, {
+        tenantId: scope,
+        provider: args.provider!,
+        externalKey: args.externalKey!,
+      });
+      if (result === "notFound") throw new CommandError("notFound", "연결된 directory identity를 찾을 수 없습니다");
       return { data: { provider: args.provider, externalKey: args.externalKey }, message: "directory identity 연결 해제 완료" };
     },
   },
diff --git a/packages/db/src/migrations/029_identity_claims.ts b/packages/db/src/migrations/029_identity_claims.ts
index 8f93975..0ef93c7 100644
--- a/packages/db/src/migrations/029_identity_claims.ts
+++ b/packages/db/src/migrations/029_identity_claims.ts
@@ -10,12 +10,14 @@ import type { Migration } from "../migrate.ts";
  * 판정이라, "묘비를 읽은 뒤 연결을 넣는" 사이에 해제가 끼어 재사용되는 경합이 없다(2026-10-19 리뷰). 원래 주인에게 다시 잇는
  * 것은 허용한다(실수로 끊은 것을 되돌릴 수 있어야 한다).
  *
- * - `realm`: "admin"(관리 콘솔 관리자, `admin_identities`) 또는 "account"(메일 계정, `directory_identities`의 `oidc:` 행).
+ * - `realm`: "admin"(관리 콘솔 관리자), "account"(메일 계정 OIDC), "directory"(LDAP/AD 외부 키 묘비).
  * - `provider`: 설정의 공급자 id다. issuer는 공급자 설정이 고정하고, 같은 영역에서 같은 issuer를 두 id로 쓰는 설정은 기동이
  *   거부한다(`oidc-config.ts`). 그래서 결정의 "iss"는 이것으로 대신한다. ⚠ 공급자 id를 바꾸거나 다른 IdP에 다시 쓰지 말 것.
  * - account 영역도 테넌트와 무관하게 전역이다 — 공급자의 테넌트를 옮겨도 "iss+sub는 하나"라는 결정을 그대로 지킨다.
  * - 개인정보가 아니라 IdP의 불투명 식별자뿐이다 — 탈퇴 뒤에도 남겨 둔다(같은 결정).
- * - LDAP/AD 디렉터리 연결(`directory-identity-*`)은 대상이 아니다: SSO 식별자가 아니고, objectGUID는 IdP가 재활용하지 않는다.
+ * - 2026-10-19부터 LDAP/AD 디렉터리 연결도 같은 표에 realm="directory"로 남긴다. 계정 드레인·연결 해제 뒤
+ *   다음 동기화가 같은 외부 키를 다른 계정에 잇는 것을 막는다. 외부 키 원문 대신 SHA-256만 저장한다.
+ *   029 당시 연결은 별도 backfill하지 않는다 — 해제·드레인이 행을 지우는 같은 배치에서 먼저 기록한다.
  *
  * 이미 있는 연결은 그 주인으로 옮겨 적는다(account 영역은 테넌트가 달라도 같은 sub면 하나만 — 가장 작은 계정 id).
  * 마이그레이션은 문장 단위로 다시 돌 수 있으므로 옮겨 적기는 NOT EXISTS로 멱등이다.
diff --git a/packages/store/src/account-drain.ts b/packages/store/src/account-drain.ts
index aa19e68..6a5eba4 100644
--- a/packages/store/src/account-drain.ts
+++ b/packages/store/src/account-drain.ts
@@ -18,11 +18,13 @@
  * - 서버 세 대의 리퍼가 같은 계정을 함께 지울 수 있다. 문장이 전부 멱등해 결과는 같다.
  *
  * 지우지 않는 것:
- * - `identity_claims`(SSO sub의 최초 주인) — 같은 sub를 다른 계정에 다시 잇지 못하게 하는 기록이라
- *   계정이 사라져도 남겨야 한다(마이그레이션 029). 값은 IdP의 불투명 식별자다.
+ * - `identity_claims`(SSO sub·디렉터리 외부 키의 최초 주인) — 같은 식별자를 다른 계정에 다시 잇지 못하게 하는
+ *   묘비라 계정이 사라져도 남긴다(마이그레이션 029). 디렉터리 외부 키는 SHA-256만 적는다.
  * - 블롭 파일 — `blob_refs`를 지우면 블롭 GC가 참조 0으로 보고 회수한다. GC가 `off`면 파일은 남는다.
  */
 import { ACCOUNT_STATUS, MTA_QUEUE_STATUS, REF_KIND, type DbDriver, type Statement } from "@ionosphere/db";
+import { chunk } from "./chunk.ts";
+import { directoryClaimStatement } from "./directory-claims.ts";
 
 /** 계정 드레인 마감 — 회원 공통 기준(2026-10-19 P8). 넘기면 지표가 overdue로 센다. */
 export const ACCOUNT_DRAIN_DEADLINE_MS = 24 * 60 * 60 * 1000;
@@ -100,7 +102,23 @@ export async function drainAccount(db: DbDriver, accountId: string, opts: { maxP
    * 배달 결과를 쓸 곳이 사라진다. 리스는 짧아서 다음 tick이면 풀린다. 남아 있으면 accounts 행을 지우지 않는다.
    * 리스가 **만료된** 발송 중 행은 지운다 — 그대로 두면 워커가 회수해 탈퇴한 계정의 메일을 다시 보내려 한다.
    */
-  const results = await db.batch([...accountStatements(accountId, opts.now ?? Date.now()), finalDeleteStatement(accountId)]);
+  const { rows: directoryLinks } = await db.query({
+    sql: "SELECT tenant_id, provider, external_key FROM directory_identities WHERE account_id = ? AND provider NOT LIKE 'oidc:%'",
+    params: [accountId],
+  });
+  const now = opts.now ?? Date.now();
+  /**
+   * D1은 배치당 문장 1000개가 상한이다. 묘비를 900개씩 먼저 남긴다 — 계정은 이미 삭제 중이고
+   * 되살릴 수 없으므로 중간에 죽어도 다음 tick이 같은 INSERT IGNORE를 다시 하며 수렴한다.
+   */
+  for (const part of chunk(directoryLinks, 900)) {
+    await db.batch(
+      part.map((row) =>
+        directoryClaimStatement(db, String(row.tenant_id), String(row.provider), String(row.external_key), accountId, now),
+      ),
+    );
+  }
+  const results = await db.batch([...accountStatements(accountId, now), finalDeleteStatement(accountId)]);
   return { messages, done: (results[results.length - 1]?.changes ?? 0) > 0 };
 }
 
diff --git a/packages/store/src/directory-claims.ts b/packages/store/src/directory-claims.ts
new file mode 100644
index 0000000..19899f3
--- /dev/null
+++ b/packages/store/src/directory-claims.ts
@@ -0,0 +1,243 @@
+/**
+ * 디렉터리(LDAP/AD) 연결의 최초 주인 기록 — SSO sub와 같은 규칙을 디렉터리 외부 키에도 건다(2026-10-19 사용자 결정).
+ *
+ * 왜: 계정 드레인(SCHEMA §7-7)이 탈퇴한 계정의 `directory_identities` 행을 지운다. 그러면 다음 동기화가 같은 외부 키를
+ * "연결 없음"으로 다시 만들고, 운영자가 그 키를 **다른 계정**에 이을 수 있었다. 해제(`directory-identity-unlink`)도 같다.
+ * 예전 사람의 디렉터리 로그인이 새 계정으로 들어오는 길이다. 그래서 처음 연결할 때 주인을 `identity_claims`에 남기고
+ * 지우지 않는다(마이그레이션 029의 표를 그대로 쓴다, realm="directory").
+ *
+ * 연결은 한 원자 배치다: 기록을 INSERT IGNORE하고(최초 주인이 이긴다), `directory_identities` UPDATE에 「주인이 나」
+ * 조건을 붙인다. 같은 계정이 같은 공급자의 두 키를 동시에 claim해 UNIQUE 충돌이 나도 claim까지 함께 롤백된다.
+ *
+ * 이 변경 전에 이미 있던 연결은 기록이 없다. 해제 명령은 행을 지우는 같은 배치에서, 계정 드레인은 데이터 삭제 전에
+ * 청크로 그 주인을 먼저 기록한다. 그래서 옮겨 적는 마이그레이션이 필요 없다(방언마다 문자열 연결·해시가 달라 SQL로는 같은 키를 못 만든다).
+ *
+ * 키 모양: `provider` 열에 `<테넌트 id>:<공급자>`(26+1+32 ≤ 64자), `subject` 열에 외부 키의 SHA-256.
+ * 원문 키는 512자이고 DN처럼 개인정보를 담을 수 있어 묘비에는 SHA-256만 남긴다. 이는 가명화라 후보 DN을
+ * 아는 사람의 대조까지 막지는 않는다. 서버 비밀 HMAC은 키 회전 때 묘비를 무효화하므로 쓰지 않는다.
+ */
+import { sha256hex, ulid } from "@ionosphere/core";
+import { BatchConflictError, type DbDriver, type Statement } from "@ionosphere/db";
+import { directoryMembershipSource } from "./directory-sync.ts";
+
+export const DIRECTORY_CLAIM_REALM = "directory";
+
+function directoryClaimKey(tenantId: string, provider: string, externalKey: string): { provider: string; subject: string } {
+  return { provider: `${tenantId}:${provider}`, subject: `sha256:${sha256hex(externalKey)}` };
+}
+
+/** 주인 기록 — 이미 있으면 그대로 둔다(최초 주인이 이긴다). 해제·계정 드레인도 이 문장을 삭제 앞에 둔다. */
+export function directoryClaimStatement(
+  db: DbDriver,
+  tenantId: string,
+  provider: string,
+  externalKey: string,
+  ownerId: string,
+  now: number,
+): Statement {
+  const k = directoryClaimKey(tenantId, provider, externalKey);
+  return {
+    sql: db.insertIgnore("identity_claims", ["realm", "provider", "subject", "owner_id", "claimed_at"]),
+    params: [DIRECTORY_CLAIM_REALM, k.provider, k.subject, ownerId, now],
+  };
+}
+
+export type DirectoryLinkResult = "ok" | "conflict" | "retired";
+
+export interface DirectoryLinkInput {
+  tenantId: string;
+  provider: string;
+  externalKey: string;
+  accountId: string;
+  displayName: string | null;
+  now?: number;
+}
+
+const DIRECTORY_LINK_RETRY_LIMIT = 5;
+
+/**
+ * 외부 키를 계정에 연결하고 최초 주인을 남긴다.
+ *
+ * `retired`: 이 키가 예전에 다른 계정에 연결됐었다(현재 행이 끊겨 있어도 묘비가 막음).
+ * `conflict`: 지금 다른 계정에 연결돼 있거나, 이 계정이 같은 공급자의 다른 키에 연결돼 있다.
+ *
+ * MySQL은 서로 다른 identity 행을 같은 계정에 동시에 연결하면 두 트랜잭션이 계정 유니크 인덱스를
+ * 반대 순서로 기다려 교착될 수 있다. DB가 한쪽을 되돌린 뒤 **전체 판정부터** 다시 해야 한다 — 실패한
+ * batch만 그대로 재전송하면 그사이 이긴 연결·claim을 보지 못한다. PG의 deadlock/serialization과
+ * SQLite busy도 같은 규칙으로 처리한다.
+ */
+export async function linkDirectoryIdentityWithClaim(db: DbDriver, input: DirectoryLinkInput): Promise<DirectoryLinkResult> {
+  let lastError: unknown;
+  for (let attempt = 0; attempt < DIRECTORY_LINK_RETRY_LIMIT; attempt++) {
+    try {
+      return await linkDirectoryIdentityAttempt(db, input);
+    } catch (error) {
+      if (!isRetryableTransactionError(error)) throw error;
+      lastError = error;
+      if (attempt < DIRECTORY_LINK_RETRY_LIMIT - 1) {
+        await new Promise((resolve) => setTimeout(resolve, 5 * (attempt + 1) + Math.floor(Math.random() * 10)));
+      }
+    }
+  }
+  throw lastError;
+}
+
+function isRetryableTransactionError(error: unknown): boolean {
+  if (!error || typeof error !== "object") return false;
+  // node:sqlite는 code="ERR_SQLITE_ERROR", errcode=5로 busy를 표현한다.
+  if ("errcode" in error && Number((error as { errcode: unknown }).errcode) === 5) return true;
+  if (!("code" in error)) return false;
+  const code = String((error as { code: unknown }).code);
+  return code === "ER_LOCK_DEADLOCK" || code === "ER_LOCK_WAIT_TIMEOUT" || code === "40P01" || code === "40001";
+}
+
+async function linkDirectoryIdentityAttempt(db: DbDriver, input: DirectoryLinkInput): Promise<DirectoryLinkResult> {
+  if (input.provider.startsWith("oidc:")) throw new Error("OIDC identity는 OIDC 연결 함수로 다뤄야 합니다");
+  const now = input.now ?? Date.now();
+  const k = directoryClaimKey(input.tenantId, input.provider, input.externalKey);
+  const state = async (): Promise<{ accountId: string | null; ownerId: string | null }> => {
+    const { rows } = await db.query({
+      sql: `SELECT d.account_id, c.owner_id
+              FROM directory_identities d LEFT JOIN identity_claims c
+                ON c.realm = ? AND c.provider = ? AND c.subject = ?
+             WHERE d.tenant_id = ? AND d.provider = ? AND d.external_key = ?`,
+      params: [DIRECTORY_CLAIM_REALM, k.provider, k.subject, input.tenantId, input.provider, input.externalKey],
+    });
+    return {
+      accountId: rows[0]?.account_id == null ? null : String(rows[0].account_id),
+      ownerId: rows[0]?.owner_id == null ? null : String(rows[0].owner_id),
+    };
+  };
+
+  const before = await state();
+  if (before.accountId !== null && before.accountId !== input.accountId) return "conflict";
+  if (before.ownerId !== null && before.ownerId !== input.accountId) return "retired";
+
+  try {
+    const statements: Statement[] = [
+      {
+        sql: `UPDATE directory_identities SET account_id = ?
+               WHERE tenant_id = ? AND provider = ? AND external_key = ? AND status = 1
+                 AND (account_id IS NULL OR account_id = ?)
+                 AND EXISTS (SELECT 1 FROM accounts WHERE id = ? AND tenant_id = ? AND status = 1)
+                 AND NOT EXISTS (SELECT 1 FROM identity_claims
+                                  WHERE realm = ? AND provider = ? AND subject = ? AND owner_id <> ?)`,
+        params: [
+          input.accountId,
+          input.tenantId,
+          input.provider,
+          input.externalKey,
+          input.accountId,
+          input.accountId,
+          input.tenantId,
+          DIRECTORY_CLAIM_REALM,
+          k.provider,
+          k.subject,
+          input.accountId,
+        ],
+      },
+    ];
+    if (before.ownerId === null) {
+      /**
+       * 연결 UPDATE가 실제 승인된 경우에만 claim한다. 삭제가 사전 조회와 배치 사이에 시작되면 UPDATE도 SELECT도
+       * 0행이라 묘비가 생기지 않는다. 동시 최초 연결은 identity 행 잠금 뒤 한쪽만 account_id를 얻고, 진 쪽의
+       * INSERT SELECT는 0행이다. 같은 계정의 두 키가 UNIQUE 충돌하면 이 INSERT까지 한 배치로 롤백된다.
+       */
+      statements.push({
+        sql: `INSERT INTO identity_claims (realm, provider, subject, owner_id, claimed_at)
+              SELECT ?, ?, ?, ?, ? FROM directory_identities d JOIN accounts a ON a.id = d.account_id
+               WHERE d.tenant_id = ? AND d.provider = ? AND d.external_key = ? AND d.account_id = ? AND d.status = 1
+                 AND a.tenant_id = ? AND a.status = 1`,
+        params: [
+          DIRECTORY_CLAIM_REALM,
+          k.provider,
+          k.subject,
+          input.accountId,
+          now,
+          input.tenantId,
+          input.provider,
+          input.externalKey,
+          input.accountId,
+          input.tenantId,
+        ],
+      });
+    }
+    /**
+     * identity principal은 외부 주체 자체라 account_id를 갖지 않는다. 연결이 승인된 경우에만 만들고 갱신한다.
+     * 연결·묘비·principal·권한 버전을 한 배치에 둬 중간 실패가 반쪽 상태를 남기지 않는다.
+     */
+    const linked = `EXISTS (SELECT 1 FROM directory_identities d JOIN identity_claims c
+                              ON c.realm = ? AND c.provider = ? AND c.subject = ? AND c.owner_id = d.account_id
+                            WHERE d.tenant_id = ? AND d.provider = ? AND d.external_key = ? AND d.account_id = ? AND d.status = 1)`;
+    const principalColumns = ["id", "tenant_id", "kind", "provider", "external_key", "display_name", "created_at"] as const;
+    const marker = `VALUES (${principalColumns.map(() => "?").join(", ")})`;
+    const principalInsert = db.insertIgnore("principals", principalColumns);
+    if (!principalInsert.includes(marker)) throw new Error("directory principal INSERT 변환 실패");
+    statements.push({
+      // FROM 한 행을 명시한다 — 오래된 MariaDB도 FROM 없는 SELECT … WHERE에 기대지 않게.
+      sql: principalInsert.replace(marker, `SELECT ?, ?, 0, ?, ?, ?, ? FROM (SELECT 1) AS one WHERE ${linked}`),
+      params: [
+        ulid(), input.tenantId, input.provider, input.externalKey, input.displayName, now,
+        DIRECTORY_CLAIM_REALM, k.provider, k.subject, input.tenantId, input.provider, input.externalKey, input.accountId,
+      ],
+    });
+    statements.push({
+      sql: `UPDATE principals SET display_name = ?
+             WHERE tenant_id = ? AND kind = 0 AND provider = ? AND external_key = ? AND ${linked}`,
+      params: [
+        input.displayName, input.tenantId, input.provider, input.externalKey,
+        DIRECTORY_CLAIM_REALM, k.provider, k.subject, input.tenantId, input.provider, input.externalKey, input.accountId,
+      ],
+    });
+    statements.push({
+      sql: `UPDATE accounts SET permissions_version = permissions_version + 1
+             WHERE id = ? AND tenant_id = ? AND ${linked}`,
+      params: [
+        input.accountId, input.tenantId,
+        DIRECTORY_CLAIM_REALM, k.provider, k.subject, input.tenantId, input.provider, input.externalKey, input.accountId,
+      ],
+    });
+    await db.batch(statements);
+  } catch (error) {
+    if (!(error instanceof BatchConflictError)) throw error;
+    // 같은 계정의 같은 요청이 동시에 성공해 claim UNIQUE만 경합했으면 이미 원하는 상태다.
+  }
+
+  const after = await state();
+  if (after.accountId !== null && after.accountId !== input.accountId) return "conflict";
+  if (after.ownerId !== null && after.ownerId !== input.accountId) return "retired";
+  if (after.accountId !== input.accountId || after.ownerId !== input.accountId) return "conflict";
+  return "ok";
+}
+
+/**
+ * 연결 해제 — 기존 연결을 처음 보는 경우도, 행을 끊는 **같은 배치 안에서** 최초 주인을 기록한다.
+ * claim은 지우지 않으므로 원래 계정만 다시 연결할 수 있다.
+ */
+export async function unlinkDirectoryIdentityWithClaim(
+  db: DbDriver,
+  input: { tenantId: string; provider: string; externalKey: string; now?: number },
+): Promise<"ok" | "notFound"> {
+  if (input.provider.startsWith("oidc:")) throw new Error("OIDC identity는 OIDC 연결 함수로 다뤄야 합니다");
+  const { rows } = await db.query({
+    sql: "SELECT account_id FROM directory_identities WHERE tenant_id = ? AND provider = ? AND external_key = ?",
+    params: [input.tenantId, input.provider, input.externalKey],
+  });
+  const accountId = rows[0]?.account_id == null ? null : String(rows[0].account_id);
+  if (!accountId) return "notFound";
+  await db.batch([
+    directoryClaimStatement(db, input.tenantId, input.provider, input.externalKey, accountId, input.now ?? Date.now()),
+    { sql: "DELETE FROM account_memberships WHERE account_id = ? AND source = ?", params: [accountId, directoryMembershipSource(input.provider)] },
+    // 기능 전 sync가 identity principal에 account_id를 넣었을 수 있다. 해제 뒤 외부 주체 ACL이 계정에 남지 않게 걷는다.
+    {
+      sql: "UPDATE principals SET account_id = NULL WHERE tenant_id = ? AND kind = 0 AND provider = ? AND external_key = ? AND account_id = ?",
+      params: [input.tenantId, input.provider, input.externalKey, accountId],
+    },
+    {
+      sql: "UPDATE directory_identities SET account_id = NULL WHERE tenant_id = ? AND provider = ? AND external_key = ? AND account_id = ?",
+      params: [input.tenantId, input.provider, input.externalKey, accountId],
+    },
+    { sql: "UPDATE accounts SET permissions_version = permissions_version + 1 WHERE id = ? AND tenant_id = ?", params: [accountId, input.tenantId] },
+  ]);
+  return "ok";
+}
diff --git a/packages/store/src/directory-sync.ts b/packages/store/src/directory-sync.ts
index 4966157..82d9f37 100644
--- a/packages/store/src/directory-sync.ts
+++ b/packages/store/src/directory-sync.ts
@@ -1,6 +1,6 @@
 import { createHash } from "node:crypto";
 import { ulid, type DirectorySnapshot } from "@ionosphere/core";
-import type { DbDriver, Statement } from "@ionosphere/db";
+import { ACCOUNT_STATUS, type DbDriver, type Statement } from "@ionosphere/db";
 
 // MySQL utf8mb4의 3072바이트 PK 한도 때문에 원문 키는 보존하고 PK에는 고정 길이 digest를 쓴다.
 function externalKeyHash(value: string): string {
@@ -18,7 +18,6 @@ export interface DirectoryIdentitySync {
   loginNames: readonly string[];
   email: string | null;
   displayName: string | null;
-  accountId: string | null;
   groupExternalKeys: readonly string[];
 }
 
@@ -35,38 +34,56 @@ export interface DirectorySyncInput {
   groups: readonly DirectoryGroupSync[];
 }
 
-/** 외부 snapshot에는 로컬 id를 섞지 않고, 기존 immutable-key 연결만 DB에서 복원한다. */
-export async function directorySyncInputFromSnapshot(
-  db: DbDriver,
+/**
+ * 외부 snapshot에는 로컬 account id를 섞지 않는다. 연결은 `directory-identity-link`만 바꾸고,
+ * sync는 배치가 실행되는 순간 DB의 현재 연결을 읽는다. 예전에는 여기서 account_id를 미리 읽어
+ * input에 넣었고, 그 뒤 관리자가 해제해도 오래된 input이 연결·그룹 권한을 되살렸다(2026-10-19 리뷰).
+ */
+export function directorySyncInputFromSnapshot(
   input: { tenantId: string; provider: string; now: number; snapshot: DirectorySnapshot },
-): Promise<DirectorySyncInput> {
-  const existing = await db.query({
-    sql: "SELECT external_key, account_id FROM directory_identities WHERE tenant_id = ? AND provider = ?",
-    params: [input.tenantId, input.provider],
-  });
-  const accountByExternalKey = new Map(existing.rows.map((row) => [String(row.external_key), row.account_id == null ? null : String(row.account_id)]));
+): DirectorySyncInput {
   return {
     tenantId: input.tenantId,
     provider: input.provider,
     now: input.now,
-    identities: input.snapshot.identities.map((identity) => ({
-      ...identity,
-      accountId: accountByExternalKey.get(identity.externalKey) ?? null,
-    })),
+    identities: input.snapshot.identities.map((identity) => ({ ...identity })),
     groups: input.snapshot.groups.map((group) => ({ externalKey: group.externalKey, memberExternalKeys: group.memberExternalKeys })),
   };
 }
 
+/**
+ * `insertIgnore()`의 VALUES 자리에 조건부 SELECT를 넣는다. 네 방언 모두 같은 골격이다:
+ * SQLite/D1 `INSERT OR IGNORE`, PG `… ON CONFLICT DO NOTHING`, MySQL `INSERT IGNORE`.
+ *
+ * 왜: account id를 배치 전에 읽어 VALUES로 넣으면 그 사이의 해제·삭제를 되돌린다. SELECT가 배치 시점의
+ * `directory_identities`와 `accounts.status`를 다시 봐야 한다. 별도 다이얼렉트 분기는 만들지 않는다.
+ */
+function insertIgnoreSelect(
+  db: DbDriver,
+  table: string,
+  columns: readonly string[],
+  select: string,
+  params: readonly unknown[],
+): Statement {
+  const marker = `VALUES (${columns.map(() => "?").join(", ")})`;
+  const base = db.insertIgnore(table, columns);
+  if (!base.includes(marker)) throw new Error(`insertIgnore SELECT 변환 실패: ${table}`);
+  return { sql: base.replace(marker, select), params };
+}
+
 /**
  * 디렉터리 snapshot을 한 배치로 반영한다. 외부 조회가 실패한 경우 호출하지 않으며,
- * snapshot에 없는 행을 지우지 않는 이유는 일시적인 LDAP 장애가 기존 권한을 삭제하면
- * 안 되기 때문이다. 변경된 계정의 permissions_version은 같은 원자 배치에서 올린다.
+ * snapshot에 없는 행을 지우지 않는 이유는 일시적인 LDAP 장애가 기존 identity를 삭제하면 안 되기 때문이다.
+ *
+ * ★연결(account_id)은 절대 쓰지 않는다. metadata만 갱신하고 principal·membership은 이 배치 시점의 현재 연결에서
+ * 조건부 SELECT로 만든다. 그래서 sync 입력을 만든 뒤 unlink/account-delete가 끼어도 연결과 권한이 되살아나지 않는다.
  */
 export async function syncDirectorySnapshot(db: DbDriver, input: DirectorySyncInput): Promise<void> {
   if (!input.tenantId || !input.provider || !Number.isInteger(input.now)) throw new Error("directory sync 입력이 유효하지 않음");
   const membershipSource = directoryMembershipSource(input.provider);
   const statements: Statement[] = [];
   const affectedAccountIds = new Set<string>();
+
   // legacy source="directory"도 principal.provider로 범위를 좁혀 이 provider의 첫 동기화에서만 회수한다.
   const existingMemberships = await db.query({
     sql: `SELECT am.account_id, am.principal_id FROM account_memberships am
@@ -78,61 +95,102 @@ export async function syncDirectorySnapshot(db: DbDriver, input: DirectorySyncIn
     affectedAccountIds.add(String(row.account_id));
     statements.push({ sql: "DELETE FROM account_memberships WHERE account_id = ? AND principal_id = ?", params: [row.account_id, row.principal_id] });
   }
-  for (const identity of input.identities) if (identity.accountId) affectedAccountIds.add(identity.accountId);
+  const linkedAccounts = await db.query({
+    sql: `SELECT d.account_id FROM directory_identities d JOIN accounts a ON a.id = d.account_id
+           WHERE d.tenant_id = ? AND d.provider = ? AND d.account_id IS NOT NULL AND a.status != ${ACCOUNT_STATUS.deleting}`,
+    params: [input.tenantId, input.provider],
+  });
+  for (const row of linkedAccounts.rows) affectedAccountIds.add(String(row.account_id));
+
   statements.push({ sql: "DELETE FROM directory_group_members WHERE tenant_id = ? AND provider = ?", params: [input.tenantId, input.provider] });
   const groupKeys = [...new Set(input.identities.flatMap((identity) => identity.groupExternalKeys))];
   const groupPrincipalIds = new Map<string, string>();
   if (groupKeys.length > 0) {
-    const existing = await db.query({ sql: `SELECT external_key, id FROM principals WHERE tenant_id = ? AND kind = 1 AND provider = ?`, params: [input.tenantId, input.provider] });
+    const existing = await db.query({
+      sql: "SELECT external_key, id FROM principals WHERE tenant_id = ? AND kind = 1 AND provider = ?",
+      params: [input.tenantId, input.provider],
+    });
     for (const row of existing.rows) groupPrincipalIds.set(String(row.external_key), String(row.id));
   }
 
   for (const identity of input.identities) {
     const identityId = ulid();
+    // 새 identity는 무조건 연결 없음(NULL)으로 만든다. 연결은 관리 명령만 할 수 있다.
     statements.push({
-      sql: db.insertIgnore("directory_identities", ["id", "tenant_id", "provider", "external_key", "account_id", "login_names", "email", "display_name", "last_seen_at", "status"]).replace(/\?\)$/, "1)"),
-      params: [identityId, input.tenantId, input.provider, identity.externalKey, identity.accountId, JSON.stringify(identity.loginNames), identity.email, identity.displayName, input.now],
+      sql: db.insertIgnore("directory_identities", ["id", "tenant_id", "provider", "external_key", "account_id", "login_names", "email", "display_name", "last_seen_at", "status"]),
+      params: [identityId, input.tenantId, input.provider, identity.externalKey, null, JSON.stringify(identity.loginNames), identity.email, identity.displayName, input.now, 1],
     });
     statements.push({
-      sql: `UPDATE directory_identities SET account_id = ?, login_names = ?, email = ?, display_name = ?, last_seen_at = ?, status = 1
-        WHERE tenant_id = ? AND provider = ? AND external_key = ?`,
-      params: [identity.accountId, JSON.stringify(identity.loginNames), identity.email, identity.displayName, input.now, input.tenantId, input.provider, identity.externalKey],
+      sql: `UPDATE directory_identities SET login_names = ?, email = ?, display_name = ?, last_seen_at = ?, status = 1
+             WHERE tenant_id = ? AND provider = ? AND external_key = ?`,
+      params: [JSON.stringify(identity.loginNames), identity.email, identity.displayName, input.now, input.tenantId, input.provider, identity.externalKey],
     });
-    if (identity.accountId) {
-      statements.push({
-        sql: db.insertIgnore("principals", ["id", "tenant_id", "kind", "account_id", "provider", "external_key", "display_name", "created_at"]).replace("?, ?, ?,", "?, ?, 0,"),
-        params: [ulid(), input.tenantId, identity.accountId, input.provider, identity.externalKey, identity.displayName, input.now],
-      });
-      statements.push({ sql: "UPDATE principals SET account_id = ?, display_name = ? WHERE tenant_id = ? AND kind = 0 AND provider = ? AND external_key = ?", params: [identity.accountId, identity.displayName, input.tenantId, input.provider, identity.externalKey] });
-    }
-    for (const groupExternalKey of identity.groupExternalKeys) {
+
+    /**
+     * identity principal은 외부 주체 자체라 account_id를 갖지 않는다. 계정은 이미 `local` principal 하나가
+     * `(tenant_id, account_id)` 유니크를 차지한다. 여기서 account_id를 넣으면 그 제약과 충돌하고, 해제 뒤
+     * 오래된 sync가 계정 연결을 되살릴 수도 있다. 계정의 디렉터리 권한은 아래 그룹 membership이 맡는다.
+     */
+    statements.push({
+      sql: db.insertIgnore("principals", ["id", "tenant_id", "kind", "provider", "external_key", "display_name", "created_at"]),
+      params: [ulid(), input.tenantId, 0, input.provider, identity.externalKey, identity.displayName, input.now],
+    });
+    statements.push({
+      sql: "UPDATE principals SET display_name = ? WHERE tenant_id = ? AND kind = 0 AND provider = ? AND external_key = ?",
+      params: [identity.displayName, input.tenantId, input.provider, identity.externalKey],
+    });
+
+    for (const groupExternalKey of new Set(identity.groupExternalKeys)) {
       const groupPrincipalId = groupPrincipalIds.get(groupExternalKey) ?? ulid();
       groupPrincipalIds.set(groupExternalKey, groupPrincipalId);
       statements.push({
-        sql: db.insertIgnore("principals", ["id", "tenant_id", "kind", "provider", "external_key", "display_name", "created_at"]).replace("?, ?, ?,", "?, ?, 1,"),
-        params: [groupPrincipalId, input.tenantId, input.provider, groupExternalKey, groupExternalKey, input.now],
+        sql: db.insertIgnore("principals", ["id", "tenant_id", "kind", "provider", "external_key", "display_name", "created_at"]),
+        params: [groupPrincipalId, input.tenantId, 1, input.provider, groupExternalKey, groupExternalKey, input.now],
       });
-      if (identity.accountId) {
-        statements.push({
-          sql: db.insertIgnore("account_memberships", ["account_id", "principal_id", "source", "created_at"]),
-          params: [identity.accountId, groupPrincipalId, membershipSource, input.now],
-        });
-      }
+      statements.push(
+        insertIgnoreSelect(
+          db,
+          "account_memberships",
+          ["account_id", "principal_id", "source", "created_at"],
+          `SELECT d.account_id, ?, ?, ? FROM directory_identities d JOIN accounts a ON a.id = d.account_id
+            WHERE d.tenant_id = ? AND d.provider = ? AND d.external_key = ? AND d.account_id IS NOT NULL
+              AND a.status != ${ACCOUNT_STATUS.deleting}`,
+          [groupPrincipalId, membershipSource, input.now, input.tenantId, input.provider, identity.externalKey],
+        ),
+      );
     }
   }
 
   for (const group of input.groups) {
     for (const member of group.memberExternalKeys) {
       statements.push({
-        sql: `${db.insertIgnore("directory_group_members", ["tenant_id", "provider", "group_external_key", "member_external_key", "group_external_hash", "member_external_hash", "last_seen_at"])}`,
+        sql: db.insertIgnore("directory_group_members", ["tenant_id", "provider", "group_external_key", "member_external_key", "group_external_hash", "member_external_hash", "last_seen_at"]),
         params: [input.tenantId, input.provider, group.externalKey, member, externalKeyHash(group.externalKey), externalKeyHash(member), input.now],
       });
-      statements.push({ sql: "UPDATE directory_group_members SET last_seen_at = ? WHERE tenant_id = ? AND provider = ? AND group_external_key = ? AND member_external_key = ?", params: [input.now, input.tenantId, input.provider, group.externalKey, member] });
+      statements.push({
+        sql: "UPDATE directory_group_members SET last_seen_at = ? WHERE tenant_id = ? AND provider = ? AND group_external_key = ? AND member_external_key = ?",
+        params: [input.now, input.tenantId, input.provider, group.externalKey, member],
+      });
     }
   }
 
+  // 사전 조회 뒤 새로 연결된 계정도 이 snapshot으로 membership이 생기므로, 배치 시점의 현재 연결을 다시 보고 버전을 올린다.
+  statements.push({
+    sql: `UPDATE accounts SET permissions_version = permissions_version + 1
+           WHERE tenant_id = ? AND status != ${ACCOUNT_STATUS.deleting}
+             AND id IN (SELECT account_id FROM directory_identities
+                         WHERE tenant_id = ? AND provider = ? AND account_id IS NOT NULL)`,
+    params: [input.tenantId, input.tenantId, input.provider],
+  });
+  // 반대로 사전 조회 뒤 해제된 계정은 membership 삭제를 캐시에 반영해야 하므로 개별 버전도 올린다.
   for (const accountId of affectedAccountIds) {
-    statements.push({ sql: "UPDATE accounts SET permissions_version = permissions_version + 1 WHERE id = ? AND tenant_id = ?", params: [accountId, input.tenantId] });
+    statements.push({
+      sql: `UPDATE accounts SET permissions_version = permissions_version + 1
+             WHERE id = ? AND tenant_id = ? AND status != ${ACCOUNT_STATUS.deleting}
+               AND id NOT IN (SELECT account_id FROM directory_identities
+                               WHERE tenant_id = ? AND provider = ? AND account_id IS NOT NULL)`,
+      params: [accountId, input.tenantId, input.tenantId, input.provider],
+    });
   }
   await db.batch(statements);
 }
diff --git a/packages/store/src/index.ts b/packages/store/src/index.ts
index 8ee2aa4..89a424a 100644
--- a/packages/store/src/index.ts
+++ b/packages/store/src/index.ts
@@ -95,6 +95,7 @@ export { scramSegment, buildScramSegment, scramKeysFor, scramAuthorize, type Sto
 export { AUTH_SURFACES, credentialAllowsSurface, type AuthSurface } from "./auth.ts";
 export { createBayesStore } from "./bayes-store.ts";
 export { directoryMembershipSource, directorySyncInputFromSnapshot, syncDirectorySnapshot, type DirectoryGroupSync, type DirectoryIdentitySync, type DirectorySyncInput } from "./directory-sync.ts";
+export { DIRECTORY_CLAIM_REALM, directoryClaimStatement, linkDirectoryIdentityWithClaim, unlinkDirectoryIdentityWithClaim, type DirectoryLinkResult } from "./directory-claims.ts";
 export { backfillHeaderProjection, projectHeaders, HEADER_PROJECTION_LIMITS, HEADER_PROJECTION_NAMES, type HeaderBackfillOptions, type HeaderProjection, type HeaderProjectionKind, type HeaderProjectionName } from "./header-projection.ts";
 export { getOrLoadListing, listingCacheKey, ListingCache, LISTING_CACHE_LIMITS, type ListingCacheKeyParts, type ListingCacheOptions } from "./listing-cache.ts";
 
diff --git a/packages/store/test/directory-claims.test.ts b/packages/store/test/directory-claims.test.ts
new file mode 100644
index 0000000..2fe3640
--- /dev/null
+++ b/packages/store/test/directory-claims.test.ts
@@ -0,0 +1,268 @@
+/**
+ * LDAP/AD 외부 키의 최초 주인 기록 — 끊거나 계정이 탈퇴해도 같은 키를 다른 계정에 잇지 못한다.
+ *
+ * CI의 PG·MySQL(IONOSPHERE_TEST_PG_URL / IONOSPHERE_TEST_MYSQL_URL)에서도 돈다. 최초 연결 경합과
+ * 같은 계정에 두 키를 동시에 잇는 UNIQUE 경합은 메모리 SQLite만으로 증명할 수 없다.
+ */
+import { afterEach, describe, expect, test } from "@ionosphere/testkit";
+import { ACCOUNT_STATUS, allMigrations, migrate, openMysql, openPostgres, openSqlite, type DbDriver } from "@ionosphere/db";
+import {
+  directoryMembershipSource,
+  drainAccount,
+  linkDirectoryIdentityWithClaim,
+  Store,
+  syncDirectorySnapshot,
+  unlinkDirectoryIdentityWithClaim,
+} from "../src/index.ts";
+
+const SCHEMA = "ionosphere_test_directory_claims";
+const opened: DbDriver[] = [];
+
+afterEach(async () => {
+  while (opened.length > 0) {
+    try {
+      await opened.pop()!.close();
+    } catch {
+      /* 이미 닫혔다 */
+    }
+  }
+});
+
+function track(db: DbDriver): DbDriver {
+  opened.push(db);
+  return db;
+}
+
+async function freshPostgres(base: string): Promise<DbDriver> {
+  const admin = track(await openPostgres(base));
+  await admin.query({ sql: `DROP SCHEMA IF EXISTS ${SCHEMA} CASCADE` });
+  await admin.query({ sql: `CREATE SCHEMA ${SCHEMA}` });
+  const url = new URL(base);
+  url.searchParams.set("options", `-c search_path=${SCHEMA}`);
+  const db = track(await openPostgres(url.toString()));
+  await migrate(db, allMigrations);
+  return db;
+}
+
+async function freshMysql(base: string): Promise<DbDriver> {
+  const admin = track(await openMysql(base));
+  await admin.query({ sql: `DROP DATABASE IF EXISTS \`${SCHEMA}\`` });
+  await admin.query({ sql: `CREATE DATABASE \`${SCHEMA}\` CHARACTER SET utf8mb4` });
+  const url = new URL(base);
+  url.pathname = `/${SCHEMA}`;
+  const db = track(await openMysql(url.toString()));
+  await migrate(db, allMigrations);
+  return db;
+}
+
+async function freshSqlite(): Promise<DbDriver> {
+  const db = track(await openSqlite());
+  await migrate(db, allMigrations);
+  return db;
+}
+
+const dialects: { name: string; open: (() => Promise<DbDriver>) | null }[] = [
+  { name: "sqlite", open: freshSqlite },
+  { name: "postgres", open: process.env.IONOSPHERE_TEST_PG_URL ? () => freshPostgres(process.env.IONOSPHERE_TEST_PG_URL!) : null },
+  { name: "mysql", open: process.env.IONOSPHERE_TEST_MYSQL_URL ? () => freshMysql(process.env.IONOSPHERE_TEST_MYSQL_URL!) : null },
+];
+
+async function insertIdentity(
+  db: DbDriver,
+  input: { id: string; tenantId: string; provider: string; externalKey: string; accountId?: string },
+): Promise<void> {
+  await db.batch([
+    {
+      sql: `INSERT INTO directory_identities
+              (id, tenant_id, provider, external_key, account_id, login_names, display_name, last_seen_at, status)
+            VALUES (?, ?, ?, ?, ?, '[]', ?, 1, 1)`,
+      params: [input.id, input.tenantId, input.provider, input.externalKey, input.accountId ?? null, input.externalKey],
+    },
+  ]);
+}
+
+function link(db: DbDriver, tenantId: string, provider: string, externalKey: string, accountId: string) {
+  return linkDirectoryIdentityWithClaim(db, { tenantId, provider, externalKey, accountId, displayName: externalKey, now: 1 });
+}
+
+for (const d of dialects) {
+  describe.skipIf(d.open === null)(`디렉터리 외부 키 최초 주인 — ${d.name}`, () => {
+    test("연결을 끊어도 다른 계정에는 다시 잇지 못하고 원래 계정에는 다시 잇는다", async () => {
+      const db = await d.open!();
+      const store = new Store(db);
+      const { tenantId } = await store.createTenant("t");
+      const a = await store.createAccount({ tenantId, email: "a@ionosphere.test" });
+      const b = await store.createAccount({ tenantId, email: "b@ionosphere.test" });
+      await insertIdentity(db, { id: "dir-a", tenantId, provider: "ad", externalKey: "guid:one" });
+
+      expect(await link(db, tenantId, "ad", "guid:one", a.accountId)).toBe("ok");
+      expect(await unlinkDirectoryIdentityWithClaim(db, { tenantId, provider: "ad", externalKey: "guid:one", now: 2 })).toBe("ok");
+      expect(await link(db, tenantId, "ad", "guid:one", b.accountId)).toBe("retired");
+      expect(await link(db, tenantId, "ad", "guid:one", a.accountId)).toBe("ok");
+    });
+
+    test("같은 외부 키의 동시 최초 연결은 한 계정만 주인이 된다", async () => {
+      const db = await d.open!();
+      const store = new Store(db);
+      const { tenantId } = await store.createTenant("t");
+      const accounts = await Promise.all(
+        Array.from({ length: 6 }, (_, i) => store.createAccount({ tenantId, email: `race${i}@ionosphere.test` })),
+      );
+      await insertIdentity(db, { id: "dir-race", tenantId, provider: "ad", externalKey: "guid:race" });
+
+      const results = await Promise.all(accounts.map((a) => link(db, tenantId, "ad", "guid:race", a.accountId)));
+      expect(results.filter((r) => r === "ok")).toHaveLength(1);
+      expect(results.filter((r) => r === "conflict")).toHaveLength(accounts.length - 1);
+      const { rows: claims } = await db.query({ sql: "SELECT owner_id FROM identity_claims WHERE realm = 'directory'" });
+      const { rows: identity } = await db.query({ sql: "SELECT account_id FROM directory_identities WHERE id = ?", params: ["dir-race"] });
+      expect(claims).toHaveLength(1);
+      expect(String(identity[0]!.account_id)).toBe(String(claims[0]!.owner_id));
+    });
+
+    test("사전 조회 뒤 다른 주인이 연결·해제해도 retired 연결을 남기지 않는다", async () => {
+      const db = await d.open!();
+      const store = new Store(db);
+      const { tenantId } = await store.createTenant("t");
+      const a = await store.createAccount({ tenantId, email: "owner@ionosphere.test" });
+      const b = await store.createAccount({ tenantId, email: "late@ionosphere.test" });
+      await insertIdentity(db, { id: "dir-interleave", tenantId, provider: "ad", externalKey: "guid:interleave" });
+
+      let injected = false;
+      const interleavingDb: DbDriver = {
+        dialect: db.dialect,
+        async query(stmt) {
+          const result = await db.query(stmt);
+          // B가 빈 연결·묘비를 읽은 **직후**, A가 최초 연결하고 해제한다.
+          if (!injected && stmt.sql.includes("SELECT d.account_id, c.owner_id")) {
+            injected = true;
+            expect(await link(db, tenantId, "ad", "guid:interleave", a.accountId)).toBe("ok");
+            expect(await unlinkDirectoryIdentityWithClaim(db, { tenantId, provider: "ad", externalKey: "guid:interleave", now: 2 })).toBe("ok");
+          }
+          return result;
+        },
+        async batch(stmts) {
+          return await db.batch(stmts);
+        },
+        insertIgnore(table, columns) {
+          return db.insertIgnore(table, columns);
+        },
+        async close() {
+          await db.close();
+        },
+      };
+
+      expect(await link(interleavingDb, tenantId, "ad", "guid:interleave", b.accountId)).toBe("retired");
+      const { rows } = await db.query({ sql: "SELECT account_id FROM directory_identities WHERE id = ?", params: ["dir-interleave"] });
+      expect(rows[0]!.account_id).toBe(null);
+    });
+
+    test("같은 계정에 두 키를 동시에 잇다 진 쪽은 claim도 롤백된다", async () => {
+      const db = await d.open!();
+      const store = new Store(db);
+      const { tenantId } = await store.createTenant("t");
+      const a = await store.createAccount({ tenantId, email: "one@ionosphere.test" });
+      const b = await store.createAccount({ tenantId, email: "two@ionosphere.test" });
+      await insertIdentity(db, { id: "dir-x", tenantId, provider: "ad", externalKey: "guid:x" });
+      await insertIdentity(db, { id: "dir-y", tenantId, provider: "ad", externalKey: "guid:y" });
+
+      const results = await Promise.all([link(db, tenantId, "ad", "guid:x", a.accountId), link(db, tenantId, "ad", "guid:y", a.accountId)]);
+      expect(results.filter((r) => r === "ok")).toHaveLength(1);
+      const { rows: linked } = await db.query({
+        sql: "SELECT external_key FROM directory_identities WHERE tenant_id = ? AND provider = ? AND account_id = ?",
+        params: [tenantId, "ad", a.accountId],
+      });
+      const winner = String(linked[0]!.external_key);
+      const loser = winner === "guid:x" ? "guid:y" : "guid:x";
+      // 진 쪽의 claim이 함께 롤백돼 다른 계정이 정상 연결할 수 있다.
+      expect(await link(db, tenantId, "ad", loser, b.accountId)).toBe("ok");
+      expect((await db.query({ sql: "SELECT COUNT(*) AS n FROM identity_claims WHERE realm = 'directory'" })).rows[0]!.n).toBe(2);
+    });
+
+    test("기능 도입 전 연결도 해제 직전에 claim하고, 외부 키 원문은 묘비에 남기지 않는다", async () => {
+      const db = await d.open!();
+      const store = new Store(db);
+      const { tenantId } = await store.createTenant("t");
+      const a = await store.createAccount({ tenantId, email: "legacy@ionosphere.test" });
+      const b = await store.createAccount({ tenantId, email: "new@ionosphere.test" });
+      const externalKey = "CN=Private Person,OU=People,DC=example,DC=test";
+      await insertIdentity(db, { id: "dir-legacy", tenantId, provider: "ldap", externalKey, accountId: a.accountId });
+
+      expect(await unlinkDirectoryIdentityWithClaim(db, { tenantId, provider: "ldap", externalKey, now: 2 })).toBe("ok");
+      expect(await link(db, tenantId, "ldap", externalKey, b.accountId)).toBe("retired");
+      const claim = (await db.query({ sql: "SELECT provider, subject FROM identity_claims WHERE realm = 'directory'" })).rows[0]!;
+      expect(String(claim.provider)).toBe(`${tenantId}:ldap`);
+      expect(String(claim.subject)).toMatch(/^sha256:[0-9a-f]{64}$/);
+      expect(JSON.stringify(claim).includes("Private Person")).toBe(false);
+    });
+
+    test("삭제 중인 계정에는 연결도 claim도 만들지 않는다", async () => {
+      const db = await d.open!();
+      const store = new Store(db);
+      const { tenantId } = await store.createTenant("t");
+      const a = await store.createAccount({ tenantId, email: "deleting@ionosphere.test" });
+      await insertIdentity(db, { id: "dir-deleting", tenantId, provider: "ad", externalKey: "guid:deleting" });
+      await db.batch([
+        {
+          sql: "UPDATE accounts SET status = ?, deleting_since = ? WHERE id = ?",
+          params: [ACCOUNT_STATUS.deleting, 1, a.accountId],
+        },
+      ]);
+
+      expect(await link(db, tenantId, "ad", "guid:deleting", a.accountId)).toBe("conflict");
+      expect((await db.query({ sql: "SELECT COUNT(*) AS n FROM identity_claims WHERE realm = 'directory'" })).rows[0]!.n).toBe(0);
+      expect((await db.query({ sql: "SELECT account_id FROM directory_identities WHERE id = ?", params: ["dir-deleting"] })).rows[0]!.account_id).toBe(null);
+    });
+
+    test("오래된 sync 입력은 연결 해제를 되돌리거나 그룹 membership을 되살리지 않는다", async () => {
+      const db = await d.open!();
+      const store = new Store(db);
+      const { tenantId } = await store.createTenant("t");
+      const a = await store.createAccount({ tenantId, email: "stale@ionosphere.test" });
+      await insertIdentity(db, { id: "dir-stale", tenantId, provider: "ad", externalKey: "guid:stale" });
+      expect(await link(db, tenantId, "ad", "guid:stale", a.accountId)).toBe("ok");
+      const stale = {
+        tenantId,
+        provider: "ad",
+        now: 10,
+        identities: [{ externalKey: "guid:stale", loginNames: ["stale"], email: null, displayName: "Stale", groupExternalKeys: ["guid:group"] }],
+        groups: [],
+      } as const;
+      expect(await unlinkDirectoryIdentityWithClaim(db, { tenantId, provider: "ad", externalKey: "guid:stale", now: 11 })).toBe("ok");
+      await syncDirectorySnapshot(db, stale);
+
+      const identity = await db.query({ sql: "SELECT account_id FROM directory_identities WHERE id = ?", params: ["dir-stale"] });
+      const membership = await db.query({ sql: "SELECT 1 FROM account_memberships WHERE account_id = ? AND source = ?", params: [a.accountId, directoryMembershipSource("ad")] });
+      expect(identity.rows[0]!.account_id).toBe(null);
+      expect(membership.rows).toEqual([]);
+    });
+
+    test("store 직접 호출도 oidc: 행을 우회하지 못한다", async () => {
+      const db = await d.open!();
+      const store = new Store(db);
+      const { tenantId } = await store.createTenant("t");
+      const a = await store.createAccount({ tenantId, email: "oidc@ionosphere.test" });
+      await insertIdentity(db, { id: "dir-oidc", tenantId, provider: "oidc:corp", externalKey: "sub" });
+      await expect(link(db, tenantId, "oidc:corp", "sub", a.accountId)).rejects.toThrow("OIDC identity");
+      await expect(unlinkDirectoryIdentityWithClaim(db, { tenantId, provider: "oidc:corp", externalKey: "sub" })).rejects.toThrow("OIDC identity");
+    });
+
+    test("계정 드레인도 기존 디렉터리 연결을 claim한 뒤 행을 지운다", async () => {
+      const db = await d.open!();
+      const store = new Store(db);
+      const { tenantId } = await store.createTenant("t");
+      const a = await store.createAccount({ tenantId, email: "gone@ionosphere.test" });
+      const b = await store.createAccount({ tenantId, email: "next@ionosphere.test" });
+      await insertIdentity(db, { id: "dir-gone", tenantId, provider: "ad", externalKey: "guid:gone", accountId: a.accountId });
+      await db.batch([
+        {
+          sql: "UPDATE accounts SET status = ?, deleting_since = ? WHERE id = ?",
+          params: [ACCOUNT_STATUS.deleting, 1, a.accountId],
+        },
+      ]);
+
+      expect((await drainAccount(db, a.accountId, { now: 2 })).done).toBe(true);
+      await insertIdentity(db, { id: "dir-reappeared", tenantId, provider: "ad", externalKey: "guid:gone" });
+      expect(await link(db, tenantId, "ad", "guid:gone", b.accountId)).toBe("retired");
+    });
+  });
+}
diff --git a/packages/store/test/directory-sync.test.ts b/packages/store/test/directory-sync.test.ts
index 303c0fc..3239fd5 100644
--- a/packages/store/test/directory-sync.test.ts
+++ b/packages/store/test/directory-sync.test.ts
@@ -1,18 +1,60 @@
 import { describe, expect, test } from "@ionosphere/testkit";
-import { directoryMembershipSource, directorySyncInputFromSnapshot, syncDirectorySnapshot } from "../src/directory-sync.ts";
+import { ACCOUNT_STATUS } from "@ionosphere/db";
+import { linkDirectoryIdentityWithClaim, unlinkDirectoryIdentityWithClaim } from "../src/directory-claims.ts";
+import { directoryMembershipSource, directorySyncInputFromSnapshot, syncDirectorySnapshot, type DirectorySyncInput } from "../src/directory-sync.ts";
+import { drainAccount } from "../src/account-drain.ts";
 import { setupFixture } from "./helpers.ts";
 
+function identityInput(
+  tenantId: string,
+  provider: string,
+  externalKey: string,
+  groupExternalKeys: readonly string[],
+  now: number,
+): DirectorySyncInput {
+  return {
+    tenantId,
+    provider,
+    now,
+    identities: [{ externalKey, loginNames: ["user"], email: null, displayName: "User", groupExternalKeys }],
+    groups: [],
+  };
+}
+
+/** 첫 snapshot으로 identity를 만든 뒤 관리 연결을 거친다. sync가 account_id를 직접 쓰지 않는 계약을 지킨다. */
+async function createAndLink(
+  db: Awaited<ReturnType<typeof setupFixture>>["db"],
+  input: DirectorySyncInput,
+  accountId: string,
+): Promise<void> {
+  await syncDirectorySnapshot(db, input);
+  const identity = input.identities[0]!;
+  expect(
+    await linkDirectoryIdentityWithClaim(db, {
+      tenantId: input.tenantId,
+      provider: input.provider,
+      externalKey: identity.externalKey,
+      accountId,
+      displayName: identity.displayName,
+      now: input.now,
+    }),
+  ).toBe("ok");
+}
+
 describe("directory sync", () => {
-  test("snapshot 반영과 permissions_version 증가는 한 배치로 성공한다", async () => {
+  test("snapshot metadata와 현재 연결을 바탕으로 membership·permissions_version을 한 배치로 갱신한다", async () => {
     const { db, tenantId, accountId } = await setupFixture();
-    const before = await db.query({ sql: "SELECT permissions_version FROM accounts WHERE id = ?", params: [accountId] });
-    await syncDirectorySnapshot(db, {
+    const input = {
       tenantId,
       provider: "ad",
       now: 100,
-      identities: [{ externalKey: "guid:abc", loginNames: ["user@ionosphere.test"], email: "user@ionosphere.test", displayName: "User", accountId, groupExternalKeys: ["guid:group"] }],
+      identities: [{ externalKey: "guid:abc", loginNames: ["user@ionosphere.test"], email: "user@ionosphere.test", displayName: "User", groupExternalKeys: ["guid:group"] }],
       groups: [{ externalKey: "guid:group", memberExternalKeys: ["guid:abc"] }],
-    });
+    } as const;
+    await createAndLink(db, input, accountId);
+    const before = await db.query({ sql: "SELECT permissions_version FROM accounts WHERE id = ?", params: [accountId] });
+    await syncDirectorySnapshot(db, { ...input, now: 101 });
+
     const identity = await db.query({ sql: "SELECT account_id, external_key FROM directory_identities WHERE tenant_id = ? AND provider = ?", params: [tenantId, "ad"] });
     const membership = await db.query({ sql: "SELECT source FROM account_memberships WHERE account_id = ? AND source = ?", params: [accountId, directoryMembershipSource("ad")] });
     const after = await db.query({ sql: "SELECT permissions_version FROM accounts WHERE id = ?", params: [accountId] });
@@ -25,9 +67,10 @@ describe("directory sync", () => {
 
   test("같은 snapshot 재적용은 identity와 membership을 중복 생성하지 않는다", async () => {
     const { db, tenantId, accountId } = await setupFixture();
-    const input = { tenantId, provider: "ldap", now: 200, identities: [{ externalKey: "sid:1", loginNames: ["user"], email: null, displayName: null, accountId, groupExternalKeys: ["sid:g"] }], groups: [] } as const;
-    await syncDirectorySnapshot(db, input);
+    const input = identityInput(tenantId, "ldap", "sid:1", ["sid:g"], 200);
+    await createAndLink(db, input, accountId);
     await syncDirectorySnapshot(db, { ...input, now: 201 });
+    await syncDirectorySnapshot(db, { ...input, now: 202 });
     const count = await db.query({ sql: "SELECT COUNT(*) AS count FROM directory_identities WHERE tenant_id = ?", params: [tenantId] });
     const memberships = await db.query({ sql: "SELECT COUNT(*) AS count FROM account_memberships WHERE account_id = ? AND source = ?", params: [accountId, directoryMembershipSource("ldap")] });
     expect(Number(count.rows[0]?.count)).toBe(1);
@@ -37,10 +80,11 @@ describe("directory sync", () => {
 
   test("완전 snapshot에서 그룹 제거가 membership과 권한 버전을 함께 갱신한다", async () => {
     const { db, tenantId, accountId } = await setupFixture();
-    const input = { tenantId, provider: "ad", now: 300, identities: [{ externalKey: "guid:2", loginNames: ["two"], email: null, displayName: null, accountId, groupExternalKeys: ["guid:g"] }], groups: [] } as const;
+    const input = identityInput(tenantId, "ad", "guid:2", ["guid:g"], 300);
+    await createAndLink(db, input, accountId);
     await syncDirectorySnapshot(db, input);
     const before = await db.query({ sql: "SELECT permissions_version FROM accounts WHERE id = ?", params: [accountId] });
-    await syncDirectorySnapshot(db, { ...input, now: 301, identities: [{ ...input.identities[0], groupExternalKeys: [] }] });
+    await syncDirectorySnapshot(db, { ...input, now: 301, identities: [{ ...input.identities[0]!, groupExternalKeys: [] }] });
     const memberships = await db.query({ sql: "SELECT COUNT(*) AS count FROM account_memberships WHERE account_id = ? AND source = ?", params: [accountId, directoryMembershipSource("ad")] });
     const after = await db.query({ sql: "SELECT permissions_version FROM accounts WHERE id = ?", params: [accountId] });
     expect(Number(memberships.rows[0]?.count)).toBe(0);
@@ -50,16 +94,20 @@ describe("directory sync", () => {
 
   test("provider별 source는 32자 안정 문자열이고 한 provider 동기화가 다른 membership을 지우지 않는다", async () => {
     const { db, tenantId, accountId } = await setupFixture();
-    const identity = (provider: string, group: string) => ({ tenantId, provider, now: 400, identities: [{ externalKey: `${provider}:user`, loginNames: ["user"], email: null, displayName: null, accountId, groupExternalKeys: [group] }], groups: [] });
-    await syncDirectorySnapshot(db, identity("ad-east", "ad-east:group"));
-    await syncDirectorySnapshot(db, identity("ldap-west", "ldap-west:group"));
+    const east = identityInput(tenantId, "ad-east", "ad-east:user", ["ad-east:group"], 400);
+    const west = identityInput(tenantId, "ldap-west", "ldap-west:user", ["ldap-west:group"], 400);
+    await createAndLink(db, east, accountId);
+    await syncDirectorySnapshot(db, east);
+    // 한 계정은 provider마다 identity 하나를 가질 수 있다.
+    await createAndLink(db, west, accountId);
+    await syncDirectorySnapshot(db, west);
     const eastSource = directoryMembershipSource("ad-east");
     const westSource = directoryMembershipSource("ldap-west");
     expect(eastSource.length).toBe(32);
     expect(eastSource).toBe(directoryMembershipSource("ad-east"));
     expect(eastSource === westSource).toBe(false);
 
-    await syncDirectorySnapshot(db, { ...identity("ad-east", "ad-east:group"), now: 401, identities: [{ ...identity("ad-east", "ad-east:group").identities[0]!, groupExternalKeys: [] }] });
+    await syncDirectorySnapshot(db, { ...east, now: 401, identities: [{ ...east.identities[0]!, groupExternalKeys: [] }] });
     const rows = await db.query({ sql: "SELECT source FROM account_memberships WHERE account_id = ? ORDER BY source", params: [accountId] });
     expect(rows.rows.map((row) => String(row.source))).toEqual([westSource]);
     await db.close();
@@ -67,19 +115,21 @@ describe("directory sync", () => {
 
   test("legacy directory source는 같은 provider principal만 회수한다", async () => {
     const { db, tenantId, accountId } = await setupFixture();
-    const first = { tenantId, provider: "legacy-ad", now: 500, identities: [{ externalKey: "legacy:user", loginNames: ["user"], email: null, displayName: null, accountId, groupExternalKeys: ["legacy:group"] }], groups: [] } as const;
-    await syncDirectorySnapshot(db, first);
+    const input = identityInput(tenantId, "legacy-ad", "legacy:user", ["legacy:group"], 500);
+    await createAndLink(db, input, accountId);
+    await syncDirectorySnapshot(db, input);
     await db.batch([{ sql: "UPDATE account_memberships SET source = ? WHERE account_id = ?", params: ["directory", accountId] }]);
-    await syncDirectorySnapshot(db, { ...first, now: 501, identities: [{ ...first.identities[0], groupExternalKeys: [] }] });
+    await syncDirectorySnapshot(db, { ...input, now: 501, identities: [{ ...input.identities[0]!, groupExternalKeys: [] }] });
     const rows = await db.query({ sql: "SELECT source FROM account_memberships WHERE account_id = ?", params: [accountId] });
     expect(rows.rows).toEqual([]);
     await db.close();
   });
 
-  test("snapshot 변환은 기존 externalKey의 accountId만 유지하고 새 identity를 자동 연결하지 않는다", async () => {
+  test("snapshot 변환은 외부 정보만 옮기고 로컬 accountId를 싣지 않는다", async () => {
     const { db, tenantId, accountId } = await setupFixture();
-    await syncDirectorySnapshot(db, { tenantId, provider: "ad", now: 600, identities: [{ externalKey: "guid:known", loginNames: ["old"], email: null, displayName: null, accountId, groupExternalKeys: [] }], groups: [] });
-    const converted = await directorySyncInputFromSnapshot(db, {
+    const first = identityInput(tenantId, "ad", "guid:known", [], 600);
+    await createAndLink(db, first, accountId);
+    const converted = directorySyncInputFromSnapshot({
       tenantId,
       provider: "ad",
       now: 601,
@@ -91,8 +141,47 @@ describe("directory sync", () => {
         groups: [{ externalKey: "guid:group", displayName: "Group", memberExternalKeys: ["guid:known"] }],
       },
     });
-    expect(converted.identities.map((row) => row.accountId)).toEqual([accountId, null]);
+    expect(converted.identities).toEqual([
+      { externalKey: "guid:known", loginNames: ["new"], email: "new@ionosphere.test", displayName: "Known", groupExternalKeys: ["guid:group"] },
+      { externalKey: "guid:new", loginNames: ["other"], email: null, displayName: null, groupExternalKeys: [] },
+    ]);
     expect(converted.groups).toEqual([{ externalKey: "guid:group", memberExternalKeys: ["guid:known"] }]);
     await db.close();
   });
+
+  test("오래된 sync 입력은 연결 해제를 되돌리거나 membership을 되살리지 않는다", async () => {
+    const { db, tenantId, accountId } = await setupFixture();
+    const input = identityInput(tenantId, "ad", "guid:stale", ["guid:group"], 700);
+    await createAndLink(db, input, accountId);
+    await syncDirectorySnapshot(db, input);
+    // 외부 snapshot을 읽어 입력을 만든 뒤 관리자가 연결을 끊었다.
+    const stale = { ...input, now: 701 };
+    expect(await unlinkDirectoryIdentityWithClaim(db, { tenantId, provider: "ad", externalKey: "guid:stale", now: 702 })).toBe("ok");
+    await syncDirectorySnapshot(db, stale);
+
+    const identity = await db.query({ sql: "SELECT account_id FROM directory_identities WHERE tenant_id = ? AND provider = ? AND external_key = ?", params: [tenantId, "ad", "guid:stale"] });
+    const memberships = await db.query({ sql: "SELECT 1 FROM account_memberships WHERE account_id = ? AND source = ?", params: [accountId, directoryMembershipSource("ad")] });
+    expect(identity.rows[0]?.account_id).toBe(null);
+    expect(memberships.rows).toEqual([]);
+    await db.close();
+  });
+
+  test("오래된 sync 입력은 계정 드레인 뒤 연결·principal·membership 고아를 만들지 않는다", async () => {
+    const { db, tenantId, accountId } = await setupFixture();
+    const input = identityInput(tenantId, "ad", "guid:gone", ["guid:group"], 800);
+    await createAndLink(db, input, accountId);
+    await syncDirectorySnapshot(db, input);
+    const stale = { ...input, now: 801 };
+    await db.batch([{ sql: "UPDATE accounts SET status = ?, deleting_since = ? WHERE id = ?", params: [ACCOUNT_STATUS.deleting, 1, accountId] }]);
+    expect((await drainAccount(db, accountId, { now: 802 })).done).toBe(true);
+    await syncDirectorySnapshot(db, stale);
+
+    const identity = await db.query({ sql: "SELECT account_id FROM directory_identities WHERE tenant_id = ? AND provider = ? AND external_key = ?", params: [tenantId, "ad", "guid:gone"] });
+    const principal = await db.query({ sql: "SELECT account_id FROM principals WHERE tenant_id = ? AND kind = 0 AND provider = ? AND external_key = ?", params: [tenantId, "ad", "guid:gone"] });
+    const memberships = await db.query({ sql: "SELECT 1 FROM account_memberships WHERE account_id = ?", params: [accountId] });
+    expect(identity.rows[0]?.account_id).toBe(null);
+    expect(principal.rows[0]?.account_id).toBe(null);
+    expect(memberships.rows).toEqual([]);
+    await db.close();
+  });
 });