import { pubsub } from '@things-factory/shell'
import { HOST_STEPS, type ReferenceProgress } from './reference-progress.js'
import { REFERENCE_PROGRESS_TOPIC, referenceProgressPayload } from './reference-progress-channel.js'
import { assessMaster } from './reference-assessment.js'
import { Arg, Ctx, Int, Mutation, Query, Resolver, Directive } from 'type-graphql'

import { Refusal, ScalarObject, getRepository, refusalKey } from '@things-factory/shell'

import { TwinEngine } from '../../engine/index.js'
import { readRestartPolicy } from '../../engine/restart-policy.js'
import { readBoardEquipment } from '@operato/ops-contract'

import { applyKnobDefaults, filledKnobKeys } from './knob-defaults.js'
import type { KnobDef } from './template-registry.js'
import { knobDefaultsFilled, pickOriginSite, siteEquipmentAdopted } from './reference-master.js'
import { TwinInstance } from '../twin-instance/twin-instance.js'
import { TwinReference } from './twin-reference.js'
import { runFill, seqReportOf } from './reference-fill-run.js'
import { listReferences, getReference } from './reference-registry.js'
import { getAdapter, listAdapterTypes, listAdapters } from './reference-adapter.js'
import {
  discovered, failed, detailOf,
  referenceNotFound, adapterUnknown, discoveryFailed, discoveryTimedOut
} from './discovery-result.js'
import { refuseImportSpace } from './ingest-space.js'
import { actuationAddressed, actuationResetPatch } from './actuation-verification.js'
import { refuseIntakeCollision } from './connection-portability.js'
import { startReferenceLiveFeed } from './reference-live.js'
import { listTemplates, getTemplate } from './template-registry.js'

/*
 * Face2 레퍼런스 인제스트 채널(ADR-0018) — 등록된 레퍼런스 소스를 열거하고, 골라서 트윈으로 인제스트.
 * "연결하고 내 것을 불러온다" — 엔티티는 레퍼런스에서 오지 손배선/발명이 아니다.
 */

/**
 * 발견 조회의 시간 상한 — 15초.
 *
 * 어댑터 계약에 시간 제한이 없다. 주소가 틀린 엔드포인트는 답하지 않고, 그러면 화면이 무한히 로딩된다 —
 * 사용자에게는 "실패" 보다 못한 상태다(무엇을 기다리는지 모른 채 기다린다). 실 시스템 조회가 몇 초
 * 걸리는 것은 정상이므로 넉넉히 두되, 끝은 있어야 한다.
 */
const DISCOVER_TIMEOUT_MS = 15_000
const TIMEOUT = 'discover-timeout'

function withTimeout<T>(work: Promise<T>, ms: number): Promise<T> {
  return new Promise<T>((resolve, reject) => {
    const timer = setTimeout(() => reject(Object.assign(new Error('timeout'), { code: TIMEOUT })), ms)
    work.then(
      v => { clearTimeout(timer); resolve(v) },
      e => { clearTimeout(timer); reject(e) }
    )
  })
}

/**
 * 거절을 **사용자 언어로 옮겨 보낸다** — 번역은 한 곳에서 한다.
 *
 * ── 왜 서버인가 (2026-08-14) ────────────────────────────────────────────────
 * 실측: 실행 중인 트윈을 다시 프로비저닝하면 「instance "order-check" is running — stop before
 * re-provisioning」 이라는 **영어 한 줄이 토스트로** 떴다(다섯 언어 제품에서).
 *
 * 서버와 앱은 **같은 사전**을 쓰고, 사용자의 현재 언어가 백엔드에 반영된다. 그래서 이 문구는 서버가
 * 옮기는 것이 맞다 — 프레임워크가 이미 같은 규약이다(`authenticate-401-middleware` 가
 * `context.t('error.' + errorCode, detail)` 로 옮긴다).
 *
 * ── 그러면 코드는 왜 함께 보내나 ────────────────────────────────────────────
 * 화면이 **구조로** 다뤄야 하는 거절이 따로 있다(정합성 표의 행·심각도·수리 버튼 — 그것들은
 * `twin.compose.chk.*` 처럼 화면이 옮긴다). 그리고 브라우저가 아닌 소비자(AI·스크립트·API)에게는
 * 코드가 계약이다. 문장과 코드는 서로를 대신하지 않는다.
 *
 * **두 곳에서 같은 코드를 옮기지 않는다** — 문장이 두 벌이 되면 반드시 어긋난다.
 */
function refusal(context: ResolverContext, e: any, fallback: string): Record<string, unknown> {
  const code = typeof e?.code === 'string' ? e.code : undefined
  const params = e?.params && typeof e.params === 'object' ? e.params : undefined
  /* 영어 canonical 은 항상 남긴다 — 번역 키가 없으면 그것이 보이고, 로그에는 그것이 남는다. */
  const canonical = e?.message || fallback
  /* A shell `Refusal` (a kernel refusal of the model, `asModelRefusal`) has one sentence, under `error.<code>`
     — the same key `SpeakRefusals` reads on every other road, so it is not written twice. */
  const key = e instanceof Refusal ? refusalKey(e.code) : `twin.provision.err.${code}`
  const message =
    code && typeof context.t === 'function'
      ? String(context.t(key, { ...(params ?? {}), defaultValue: canonical }))
      : canonical
  return { ok: false, error: message, ...(code ? { errorCode: code } : {}), ...(params ? { errorParams: params } : {}) }
}

/**
 * `siteEquipmentIds` 노브를 이 현장의 설비로 채운다 — **비어 있을 때만.**
 *
 * 사용자가 목록을 직접 준 경우에는 손대지 않는다(고른 것이 정본이다). 현장이 없으면 채울 것도 없다.
 * 등록부의 모델에서 읽는다(런타임이 아니다) — 트윈이 도는 중이 아니어도 그 현장의 설비는 사실이다.
 */
async function resolveSiteEquipmentKnobs(
  knobs: readonly KnobDef[] | undefined,
  effective: Record<string, any>,
  domainId: string,
  spaceId?: string
): Promise<{ key: string; count: number }[]> {
  const targets = (knobs ?? []).filter(k => k?.type === 'siteEquipment')
  if (!targets.length) return []
  const out: { key: string; count: number }[] = []
  let ids: { id: string; kind: string }[] | undefined
  for (const k of targets) {
    const given = effective[k.key]
    if (Array.isArray(given) && given.length) continue // 사용자가 고른 것이 정본이다
    if (!spaceId) {
      effective[k.key] = []
      continue
    }
    if (ids === undefined) {
      const rows = await getRepository(TwinInstance).find({ where: { domain: { id: domainId }, spaceId } })
      const seen = new Map<string, { id: string; kind: string }>()
      for (const r of rows) {
        for (const e of readBoardEquipment((r.model as any) ?? {})) {
          if (!e?.id || seen.has(String(e.id))) continue
          seen.set(String(e.id), { id: String(e.id), kind: String((e as any).kind ?? '') })
        }
      }
      ids = [...seen.values()]
    }
    effective[k.key] = ids
    out.push({ key: k.key, count: ids.length })
  }
  return out
}

@Resolver()
export class TwinReferenceResolver {
  /** 등록된 레퍼런스 소스 목록(요약 + 이미 인제스트됐는지). */
  @Query(returns => [ScalarObject], { description: 'List registered reference sources (virtual or real adapters) available to ingest.' })
  async referenceSources(@Ctx() context: ResolverContext): Promise<any[]> {
    const domainId = context.state.domain.id
    const out: any[] = []
    for (const m of listReferences()) {
      const reg = await getRepository(TwinInstance).findOne({ where: { domain: { id: domainId }, instanceId: m.source } })
      out.push({
        source: m.source,
        system: m.system,
        siteName: m.siteName,
        locationCount: m.locations.filter(l => l.role === 'location' || l.role === 'both').length,
        areaCount: m.locations.filter(l => l.role === 'area' || l.role === 'both').length,
        equipmentCount: (m.equipment ?? []).length,
        imported: !!reg,
        running: TwinEngine.owns(domainId, m.source)
      })
    }
    return out
  }

  /** 소스를 인제스트 → 공간 + 인스턴스 생성(provision). start 면 기동까지. */
  @Directive('@privilege(category: "twin", privilege: "mutation", domainOwnerGranted: true, superUserGranted: true)')
  @Mutation(returns => ScalarObject, {
    description:
      "Ingest a registered reference source into a twin (space + instance). Optionally start it. restartPolicy declares how the new twin treats its own past on reboot: 'resync' (read the source again) | 'resume' (continue its journal) | 'reset' (wipe and replay from the seed). It has no default — a twin cannot exist without the declaration."
  })
  async importFromReference(
    @Arg('source') source: string,
    @Arg('restartPolicy') restartPolicy: string,
    @Ctx() context: ResolverContext,
    @Arg('start', { nullable: true }) start?: boolean
  ): Promise<any> {
    const master = getReference(source)
    if (!master) return { ok: false, error: `unknown reference source: ${source}` }
    const domainId = context.state.domain.id
    try {
      /* 어디서 왔는지 남긴다 — 이것이 없으면 이 트윈은 두 번 다시 읽을 수 없다.
         이 경로는 **메모리에 등록된 마스터 하나**를 그대로 쓰므로 사이트 서술자가 없다(사이트 정체를
         적을 것이 없다). 다시 읽기도 같은 등록본을 집으므로 사이트를 찾을 필요가 없다. */
      const res = await TwinEngine.ingestMaster(
        domainId,
        { ...master, origin: { kind: 'reference', source } },
        readRestartPolicy(restartPolicy, 'importFromReference'),
        undefined,
        context.state.user
      )
      if (start) await TwinEngine.startFromRegistry(domainId, res.instanceId)
      return { ok: true, instanceId: res.instanceId, spaceId: res.spaceId, warnings: res.warnings }
    } catch (e: any) {
      return refusal(context, e, 'ingest failed')
    }
  }

  /**
   * **원본에서 다시 읽는다** — 이 트윈의 구조를 지금의 원본 모습으로 갱신한다.
   *
   * ── 왜 진입점이 하나여야 하나 ──────────────────────────────────────────────
   * 지금까지 다시 읽기는 **만드는 경로에 얹혀 있었다**(`importFromReference`·`createFromTemplate`).
   * 그래서 이미 있는 트윈을 갱신하려면 만들 때 무엇을 골랐는지 사람이 기억해 같은 화면을 다시
   * 거쳐야 했고, 템플릿으로 세운 트윈은 노브를 어디에도 안 남겨서 **아예 갱신할 방법이 없었다.**
   * 원천이 무엇이든 트윈 자신이 기억하게 하고(`TwinInstance.origin`), 여기 한 곳에서 그것을 따른다.
   *
   * ── 무엇이 일어나나 ────────────────────────────────────────────────────────
   * 원본에서 마스터를 다시 얻어 인제스트에 그대로 태운다. 그다음은 인제스트가 아는 일이다 —
   * 구조가 달라졌으면 새 리비전이 남고, 구조 행이 다시 그려지고, **실행 중인 미러는 멈추지 않고
   * 전환한다.** 실행 중인 시뮬레이션은 거절된다(조건이 바뀐 실험은 다른 실험이다).
   *
   * ── 못 할 때는 못 한다고 한다 ──────────────────────────────────────────────
   * `origin` 이 없는 트윈(이 자리가 생기기 전에 만들어진 것)은 다시 읽을 수 없다. 짐작으로
   * 템플릿을 고르거나 첫 레퍼런스를 집지 않는다 — 엉뚱한 원본으로 현장을 덮는 것이 훨씬 나쁘다.
   */
  /**
   * **그 번호들이 무엇이었나** — 구멍을 만났을 때 원본에 묻는다.
   *
   * `fillTwinScope` 가 「418 다음이 420」이라고 멈추면, 그 419 가 무엇이었는지 알아야 다음을
   * 정할 수 있다. 그 길이 없어서 사람이 원본 DB 를 손으로 뒤졌다.
   *
   * **판정하지 않는다.** 원본이 아는 것을 그대로 낸다 — `lost` 는 「번호는 나갔는데 행이 없다」이지
   * 「포기해도 된다」가 아니다. 그 판단은 되돌릴 수 없어서(커서가 지나가면 끝이다) 사람의 몫이다.
   */
  /*
   * **읽기인데 쓰기와 같은 문을 쓴다.** 이 문은 원본의 봉투를 그대로 냅니다 — 그 시스템이 무엇을
   * 언제 보냈는지가 드러납니다. `twinConnectionBundle` 과 같은 무게이고, 이 저장소는 그 무게를
   * 「쓰기는 GATE 하나」로 통일해 두었습니다(ADR-0027).
   */
  @Directive('@privilege(category: "twin", privilege: "mutation", domainOwnerGranted: true, superUserGranted: true)')
  @Query(returns => ScalarObject, {
    description:
      'Ask the source what it knows about a range of sequence numbers. Each number comes back as present · lost · unissued · unknown — the source reports, the human decides.'
  })
  async twinSeqReport(
    @Arg('instanceId') instanceId: string,
    @Arg('scope') scope: string,
    @Arg('from', type => Int) from: number,
    @Arg('to', type => Int) to: number,
    @Ctx() context: ResolverContext
  ): Promise<any> {
    return seqReportOf({ domainId: context.state.domain.id, instanceId, scope, from, to })
  }

  /**
   * **빠진 구간을 다시 받는다** — 훅은 반드시 놓칩니다: 우리가 내려가 있을 때, 그쪽이 못 보냈을 때.
   *
   * ── 왜 문이 필요한가 ──────────────────────────────────────────────────────
   * 번호가 비면 `takeInSequence` 가 알아챕니다. 그런데 **알아챈 뒤 그 사이를 채우는 길이 없었습니다.**
   * `fillSince` 는 2026-08-31 부터 있었고 부르는 곳이 0곳이었습니다.
   *
   * 실제로 plant 아웃박스의 1~20 을 잃고 되찾을 방법이 없는 상태를 만났습니다.
   *
   * ── `from` 은 잃은 구간을 되찾을 때만 ──────────────────────────────────────
   * 주지 않으면 커서부터 이어 받습니다. 주면 **그 번호부터 다시 묻습니다** — 이미 들어간 것은 유입
   * 경계가 되풀이로 거르므로 두 번 앉지 않습니다.
   *
   * 저장된 커서를 되돌리지 않습니다. `fillSince` 는 번호 검사를 지난 만큼만 적으므로, 되찾은 뒤
   * 커서가 원래 자리보다 앞서지 않습니다.
   */
  @Directive('@privilege(category: "twin", privilege: "mutation", domainOwnerGranted: true, superUserGranted: true)')
  @Mutation(returns => ScalarObject, {
    description:
      'Fetch the stretch this twin missed for one scope and ingest it through the same checks as the push path. Without `from` it continues from the cursor; with `from` it asks again from that number — already-ingested facts are filtered as duplicates.'
  })
  async fillTwinScope(
    @Arg('instanceId') instanceId: string,
    @Arg('scope', { description: 'The unit the source numbers by — the same name its envelopes carry (e.g. EVENTS).' }) scope: string,
    @Ctx() context: ResolverContext,
    @Arg('from', type => Int, { nullable: true, description: 'Ask again from this sequence number. Omit to continue from the cursor.' })
    from?: number,
    @Arg('maxPages', type => Int, { nullable: true, description: 'Page limit for one run. Default 50.' }) maxPages?: number
  ): Promise<any> {
    return runFill({ domainId: context.state.domain.id, instanceId, scope, from: from ?? undefined, maxPages: maxPages ?? undefined })
  }

  @Directive('@privilege(category: "twin", privilege: "mutation", domainOwnerGranted: true, superUserGranted: true)')
  @Mutation(returns => ScalarObject, {
    description:
      "Re-read a twin's structure from the source it remembers (TwinInstance.origin). Works for both reference- and template-born twins. A running mirror adopts the new structure without stopping; a running simulation is refused."
  })
  async resyncTwin(@Arg('instanceId') instanceId: string, @Ctx() context: ResolverContext): Promise<any> {
    const domainId = context.state.domain.id
    const inst = await getRepository(TwinInstance).findOne({ where: { domain: { id: domainId }, instanceId } })
    if (!inst) return { ok: false, error: `unknown twin: ${instanceId}` }

    const origin = inst.origin as any
    if (!origin?.kind)
      return {
        ok: false,
        error: `twin "${instanceId}" does not remember where it came from — it cannot be re-read. Re-create it from its template or reference so the origin is recorded.`
      }

    /*
     * 진행을 알리는 통로 — 만들 때와 **같은 것**이다. 레퍼런스에서 온 트윈만 알린다(템플릿에서 온
     * 트윈은 원본을 읽지 않으므로 알릴 단계가 없다).
     *
     * 분모는 그 어댑터가 선언한 단계 수다(`masterSteps`) + 호스트 단계. 선언하지 않은 어댑터는
     * `total` 이 없고, 그때 화면은 퍼센트를 만들지 않는다 — 모르는 진행률을 지어내지 않는다.
     */
    const progressSource: string | undefined = origin.kind === 'reference' ? origin.source : undefined
    const refRow = progressSource
      ? await getRepository(TwinReference).findOne({ where: { domain: { id: domainId }, source: progressSource } })
      : null
    const adapterForSteps = refRow ? getAdapter(refRow.adapterType) : undefined
    const declaredSteps = adapterForSteps?.masterSteps?.length
    const stepTotal = declaredSteps === undefined ? undefined : declaredSteps + HOST_STEPS.length
    const siteId = origin.siteId ?? instanceId
    let lastDone = 0
    let lastStep: ReferenceProgress['step'] = (adapterForSteps?.masterSteps?.[0] ?? 'provision') as ReferenceProgress['step']
    const emit = (p: Omit<ReferenceProgress, 'siteCount' | 'siteId' | 'siteIndex'>) => {
      if (!progressSource) return
      lastDone = p.done
      lastStep = p.step
      pubsub.publish(
        REFERENCE_PROGRESS_TOPIC,
        referenceProgressPayload(domainId, { ...p, siteId, siteIndex: 0, siteCount: 1, total: p.total ?? stepTotal, source: progressSource })
      )
    }

    try {
      const master = await this.masterFromOrigin(domainId, origin, instanceId, st =>
        emit({ step: st.key, done: st.done, total: stepTotal, detail: st.detail })
      )
      emit({ step: 'provision', done: declaredSteps ?? 0, total: stepTotal })
      /*
       * **어느 현장의 트윈이었나를 지킨다** — 다시 읽기는 구조만 새로 그리는 일이고, 현장 소속은 그때
       * 사람이 고른 것이다. 기억이 없는 옛 트윈은 **지금 있는 현장**을 그대로 쓴다(빈손으로 두면 템플릿
       * 기본 공간으로 옮겨 간다 — 알리지 않고 현장을 떠나고 그 현장의 값도 잃는다).
       */
      const keepSpace = origin.spaceId || inst.spaceId
      /* 다시 읽기는 **구조만** 새로 그린다 — 재기동 정책은 그 트윈의 선언이므로 그대로 지킨다. */
      const res = await TwinEngine.ingestMaster(
        domainId,
        master,
        readRestartPolicy(inst.restartPolicy, `resync of twin "${instanceId}"`),
        keepSpace ? { spaceId: keepSpace } : undefined,
        context.state.user
      )
      emit({ step: 'start', done: (declaredSteps ?? 0) + 1, total: stepTotal })
      /*
       * **총평도 새로 남긴다.** 다시 읽었는데 총평이 옛것이면 화면이 지난 읽기를 현재로 보여 준다 —
       * 그 표를 보고 사람이 원본을 고치러 간다.
       */
      if (refRow && progressSource) {
        const prevSpec = (refRow.scopeSpec as any) ?? {}
        await getRepository(TwinReference).update(
          { id: refRow.id },
          {
            scopeSpec: { ...prevSpec, assessments: { ...(prevSpec.assessments ?? {}), [siteId]: assessMaster(master, siteId) } },
            /*
             * ── 읽은 시각과 경고 수도 새로 남긴다 (2026-09-06) ──────────────────
             * 처음 연결하는 경로는 이 둘을 쓰는데(§`connectReference`) 다시 읽는 경로는 총평만
             * 썼다. 그래서 `lastSyncedAt` 이 「처음 트윈을 만든 시각」에 머물렀다.
             *
             * 이것이 실제로 잘못된 판단을 만들었다. 인티그레이션 레인이 커넥터를 고치고 재동기화를
             * 청한 뒤 이 칸으로 확인했는데, 값이 그대로여서 「안 돌았다」고 결론 내렸다. 실제로는
             * 돌았고 품목 106건이 들어와 있었다.
             *
             * 경고 수도 같다. 다시 읽으면서 경고가 줄거나 늘었는데 화면은 지난 수를 보인다.
             */
            lastSyncedAt: new Date().toISOString(),
            lastWarningCount: Array.isArray(res.warnings) ? res.warnings.length : null
          } as any
        )
      }
      return { ok: true, instanceId: res.instanceId, spaceId: res.spaceId, warnings: res.warnings }
    } catch (e: any) {
      /* 막힌 자리를 그대로 알린다 — 「끝낸 수」의 다음 단계다(만들 때와 같은 규칙). */
      emit({
        step: (adapterForSteps?.masterSteps?.[lastDone] ?? lastStep) as ReferenceProgress['step'],
        done: lastDone,
        total: stepTotal,
        failed: true,
        error: e?.message ?? String(e)
      })
      return refusal(context, e, 'resync failed')
    }
  }

  /** 기억해 둔 원천에서 마스터를 다시 얻는다 — 어댑터든 템플릿이든 이 한 곳에서 갈린다. */
  private async masterFromOrigin(
    domainId: string,
    origin: any,
    instanceId: string,
    /*
     * **다시 읽기도 진행을 보여야 한다** (2026-08-22).
     *
     * 만들 때는 8단계가 화면에 흐르는데 다시 읽을 때는 아무것도 없었다 — 사용자가 버튼을 누르고
     * 「진행되고 있는지 확인할 수가 없다」고 했다. 실제로 chef 는 원본 읽기가 14~21초이고 그 사이
     * 화면에 아무 변화가 없다.
     *
     * 통로는 만들 때와 **같은 것**이다(`twin-reference-progress`). 화면도 같은 표시를 쓴다 —
     * 두 벌을 만들면 한쪽만 고쳐지고 사람은 같은 일이 다르게 보이는 것을 결함으로 읽는다.
     */
    onStep?: (s: { key: any; done: number; total?: number; detail?: string }) => void
  ): Promise<any> {
    if (origin.kind === 'template') {
      const tpl = getTemplate(origin.templateId)
      if (!tpl) throw new Error(`unknown template: ${origin.templateId} — this twin cannot be rebuilt`)
      /* 저장해 둔 노브 그대로 다시 짓는다. 기본값을 다시 입히지 않는다 — 그 사이 템플릿의 기본값이
         바뀌었다면, 사용자가 고르지 않은 값이 알리지 않고 현장을 바꾸게 된다. */
      const { master } = tpl.build(origin.params ?? {})
      master.origin = origin
      return master
    }

    /* 레퍼런스 — 메모리 등록본이 있으면 그것, 없으면 저장된 어댑터 연결로 다시 가져온다.
       (부팅 뒤에는 메모리 등록본이 비어 있으므로 어댑터 경로가 실제로 쓰이는 길이다.) */
    const inMemory = getReference(origin.source)
    if (inMemory) return { ...inMemory, origin }

    const ref = await getRepository(TwinReference).findOne({ where: { domain: { id: domainId }, source: origin.source } })
    if (!ref) throw new Error(`reference "${origin.source}" is no longer registered — nothing to re-read from`)
    const adapter = getAdapter(ref.adapterType)
    if (!adapter) throw new Error(`no adapter for type: ${ref.adapterType}`)
    const cfg = ref.connectionConfig ?? {}
    const sites = await adapter.discoverSites(cfg)

    /*
     * ── 무엇을 무엇과 견주나 (2026-08-18 실측으로 고침) ────────────────────────
     *
     * 예전에는 `master.source === origin.source` 로 사이트를 찾았다. 그런데 `origin.source` 는 **레퍼런스
     * 등록 이름**이고(그 참조의 사이트 전부에 같다) `master.source` 는 **사이트별 인스턴스 id** 다 —
     * 서로 다른 것을 견줬으니 사이트가 멀쩡히 있어도 「더 이상 보고하지 않는다」로 거절했다. 사이트가
     * 여럿인 참조에서 온 트윈은 그래서 **전부** 재동기가 불가했고, 오류 문장은 원인을 참조 쪽으로
     * 잘못 가리켰다.
     *
     * 이제 ① 기억해 둔 사이트로 곧장 간다(`origin.siteId`) ② 기억이 없는 옛 트윈은 사이트를 살펴
     * **마스터의 정체(=이 인스턴스 id)** 와 견준다. 어느 쪽도 못 찾으면 **첫 사이트로 때우지 않는다** —
     * 다른 현장의 구조를 이 트윈에 덮는 것은 오류가 아니라 오류 없이 틀린 공장으로 나타난다.
     */
    const picked = pickOriginSite(sites, origin)
    if (picked.site) return { ...(await adapter.fetchMaster(cfg, picked.site, onStep)), origin }
    if (picked.reason === 'site-gone')
      throw new Error(
        `reference "${origin.source}" no longer reports site "${origin.siteId}" — refusing to overwrite this twin with a different site`
      )

    for (const site of sites) {
      const master = await adapter.fetchMaster(cfg, site)
      if (master.source === instanceId) return { ...master, origin: { ...origin, siteId: site.siteId } }
    }
    throw new Error(
      `reference "${origin.source}" does not report a site that rebuilds twin "${instanceId}" ` +
        `(this twin was imported before the site identity was recorded, and no site's master carries its id) — refusing to overwrite it with a different site`
    )
  }

  /** 등록된 트윈 템플릿(파라미터화 아키타입) 목록 — 노브 스키마 포함(클라이언트 폼용). */
  @Query(returns => [ScalarObject], { description: 'List registered twin templates (parameterized archetypes) with their knob schema.' })
  async twinTemplates(@Ctx() _context: ResolverContext): Promise<any[]> {
    return listTemplates().map(t => ({ id: t.id, label: t.label, system: t.system, description: t.description, detail: t.detail, knobs: t.knobs }))
  }

  /** 템플릿 + 노브 → 합성 마스터 → 인제스트(+기동+시나리오). 레퍼런스 import 와 같은 경로, 원천만 합성. */
  @Directive('@privilege(category: "twin", privilege: "mutation", domainOwnerGranted: true, superUserGranted: true)')
  @Mutation(returns => ScalarObject, {
    description:
      "Instantiate a twin template with params → synthetic master → ingest (+start+scenario). restartPolicy declares how the new twin treats its own past on reboot ('resync' | 'resume' | 'reset') — there is no default. Returns {ok, instanceId, spaceId}."
  })
  async createFromTemplate(
    @Arg('templateId') templateId: string,
    @Arg('params', type => ScalarObject) params: any,
    @Arg('restartPolicy') restartPolicy: string,
    @Ctx() context: ResolverContext,
    @Arg('start', { nullable: true }) start?: boolean,
    @Arg('spaceId', { nullable: true, description: 'Existing space to co-locate this twin into (add a lens to a site). Omit to create its own space.' }) spaceId?: string
  ): Promise<any> {
    const tpl = getTemplate(templateId)
    if (!tpl) return { ok: false, error: `unknown template: ${templateId}` }
    const domainId = context.state.domain.id
    try {
      /*
       * **빠진 노브를 서버가 채운다.** 기본값을 아는 곳은 노브 스키마(계약)인데 그것을 적용하는 곳이
       * 위자드뿐이었다 — 그래서 화면을 거치지 않는 호출자(AI·스크립트·API)가 파라미터를 일부만 주면
       * 템플릿 안에서 발생률이 0 이 되고 **아무것도 하지 않는 트윈**이 알리지 않고 만들어졌다.
       * 채운 노브 이름은 경고로 함께 돌려준다 — 알리지 않고 채우는 것도 알리지 않고 비우는 것만큼 위험하다.
       */
      const filled = filledKnobKeys(tpl.knobs, params)
      const effective = applyKnobDefaults(tpl.knobs, params)
      /*
       * **이 현장에 이미 있는 설비를 노브 값으로 확정한다** (`siteEquipmentIds`).
       *
       * 에너지 트윈이 「공장이 돌면 전기가 오른다」를 말하려면 부하를 **이웃 트윈의 설비 id 로**
       * 선언해야 한다(설비 하나에 사실 둘). 그 목록을 빌드할 때마다 현장에서 다시 읽으면, 사이에
       * 설비가 늘거나 준 만큼 **다시 지을 때 다른 트윈**이 된다. 그래서 여기서 한 번 해석하고
       * 그 결과를 노브 값으로 남긴다(아래 `origin.params` 에 그대로 저장된다).
       */
      const adopted = await resolveSiteEquipmentKnobs(tpl.knobs, effective, domainId, spaceId)
      const { master, scenario } = tpl.build(effective)
      /* **무엇으로 지었는지 남긴다.** 합성은 결정적이므로 같은 노브면 같은 마스터가 다시 나온다 —
         그래서 템플릿 트윈도 어댑터 트윈과 똑같이 다시 읽을 수 있다. 남기지 않으면 그 트윈은
         첫날의 모습에 영원히 머문다(원본을 잃은 트윈). */
      /* 고른 현장도 남긴다 — 없으면 다시 읽기가 트윈을 템플릿 기본 공간으로 옮겨 버린다(현장 값도 잃는다). */
      master.origin = { kind: 'template', templateId, params: effective, ...(spaceId ? { spaceId } : {}) }
      /* 공간 선택은 템플릿을 거치지 않는다 — 템플릿마다 이 노브를 알고 있어야 한다면 새 템플릿이
         알리지 않고 무시하게 된다. 합성된 마스터에 한 곳에서 얹는다(레퍼런스 인제스트와 같은 경로). */
      const res = await TwinEngine.ingestMaster(
        domainId,
        master,
        readRestartPolicy(restartPolicy, `createFromTemplate("${templateId}")`),
        { spaceId },
        context.state.user
      )
      if (start !== false) {
        await TwinEngine.startFromRegistry(domainId, res.instanceId)
        if (scenario) {
          const runtime = TwinEngine.runtime(domainId, res.instanceId)
          if (runtime) {
            runtime.scenario.load(scenario)
            runtime.scenario.start()
          }
        }
      }
      const warnings = [
        ...(filled.length ? [knobDefaultsFilled(filled)] : []),
        ...adopted.filter(a => a.count > 0).map(a => siteEquipmentAdopted(a.key, a.count)),
        ...res.warnings
      ]
      return { ok: true, instanceId: res.instanceId, spaceId: res.spaceId, warnings }
    } catch (e: any) {
      return refusal(context, e, 'template instantiation failed')
    }
  }

  /* ── 레퍼런스 관리(엔티티 기반·테넌트 스코프) — reference-management P1 ────────
   * 전역 인메모리(referenceSources/importFromReference)와 공존. 이쪽은 TwinReference 행 + 어댑터 레지스트리.
   */

  @Query(returns => [ScalarObject], { description: 'List tenant-scoped persisted twin references (adapter-backed).' })
  async twinReferences(@Ctx() context: ResolverContext): Promise<any[]> {
    const domainId = context.state.domain.id
    const rows = await getRepository(TwinReference).find({ where: { domain: { id: domainId } } })
    /*
     * provenance 라이브 대조 — 지워진 트윈은 producedCount 에서 자동 제외(상태 동기의 lazy 방식).
     *
     * **id 만 읽는다** (2026-08-21). 예전에는 인스턴스 행 전체를 읽었고, 그 행에는 **모델 JSON** 이
     * 들어 있다(자리 수백 개 + 발자국 좌표). 그래서 이 목록을 여는 데 수 메가바이트를 읽고 파싱했고,
     * 화면은 그동안 「아직 연결된 시스템이 없습니다」를 보이고 있었다 — 사용자에게는 「없다」는 답으로
     * 읽힌다. 여기서 필요한 것은 살아 있는 id 집합 하나다.
     */
    const liveIds = new Set(
      (await getRepository(TwinInstance).find({ where: { domain: { id: domainId } }, select: ['instanceId'] })).map(
        i => i.instanceId
      )
    )
    return rows.map(r => {
      const produced = (((r.scopeSpec as any)?.produced ?? []) as any[]).filter(p => liveIds.has(p.instanceId))
      return {
        source: r.source, system: r.system, siteName: r.siteName, description: r.description,
        adapterType: r.adapterType, status: r.status, lastSyncedAt: r.lastSyncedAt, lastError: r.lastError,
        producedCount: produced.length, produced,
        /*
         * **조치가 저쪽에 닿아 봤나** (ADR-0046 ②). 칸 셋은 2026-09 부터 서 있었는데 목록도 상세도
         * 이 값을 싣지 않아 **화면이 볼 길이 없었다** — 저장된 사실이 아무에게도 닿지 않았다.
         *
         * 보낼 주소가 적힌 연결에만 싣는다. 읽기만 하는 연결에 「아직 확인 안 됨」이 서면 할 일이 없는
         * 사람에게 할 일이 있는 것처럼 보인다.
         */
        ...(actuationAddressed(r.connectionConfig as any)
          ? {
              actuationState: r.actuationState ?? 'unverified',
              actuationVerifiedAt: r.actuationVerifiedAt ?? null,
              actuationFailure: r.actuationFailure ?? null
            }
          : {})
      }
    })
  }

  @Query(returns => [String], { description: 'List registered reference adapter type keys.' })
  referenceAdapterTypes(): string[] {
    return listAdapterTypes()
  }

  /**
   * 인티그레이션 그룹핑 (reference-management §2.3·D11) — 산출물을 커넥터 타입 ▸ Connection ▸ 인스턴스 2단으로.
   * 트윈-네이티브 온톨로지(integration 무의존): TwinReference 가 그룹핑 노드(source=connectionName),
   * scopeSpec.connectorType 로 상위 분류. producedCount 는 라이브 인스턴스로 대조(지워진 트윈 제외).
   */
  @Query(returns => [ScalarObject], { description: 'Group integration items 2-tier: connector type ▸ connection ▸ produced twins. (reference-management §2.3)' })
  async twinConnections(@Ctx() context: ResolverContext): Promise<any[]> {
    const domainId = context.state.domain.id
    const rows = await getRepository(TwinReference).find({ where: { domain: { id: domainId } } })
    const liveIds = new Set((await getRepository(TwinInstance).find({ where: { domain: { id: domainId } } })).map(i => i.instanceId))
    const groups = new Map<string, any[]>()
    for (const r of rows) {
      const connectorType = (r.scopeSpec as any)?.connectorType ?? r.adapterType
      const producedCount = (((r.scopeSpec as any)?.produced ?? []) as any[]).filter(p => liveIds.has(p.instanceId)).length
      const conn = { source: r.source, name: r.siteName, status: r.status, adapterType: r.adapterType, producedCount, lastSyncedAt: r.lastSyncedAt }
      if (!groups.has(connectorType)) groups.set(connectorType, [])
      groups.get(connectorType)!.push(conn)
    }
    return [...groups.entries()].map(([connectorType, connections]) => ({
      connectorType,
      connectionCount: connections.length,
      instanceCount: connections.reduce((s, c) => s + c.producedCount, 0),
      connections
    }))
  }

  /** 한 Connection 그룹 노드 상세 — 그 아래 생성된 트윈 인스턴스(라이브 대조·상태·기동여부). */
  @Query(returns => ScalarObject, { nullable: true, description: 'One connection grouping location with its produced twin instances (live-checked). (reference-management §2.3)' })
  async twinConnection(@Arg('source') source: string, @Ctx() context: ResolverContext): Promise<any | null> {
    const domainId = context.state.domain.id
    const r = await getRepository(TwinReference).findOne({ where: { domain: { id: domainId }, source } })
    if (!r) return null
    const produced = ((r.scopeSpec as any)?.produced ?? []) as any[]
    const instMap = new Map((await getRepository(TwinInstance).find({ where: { domain: { id: domainId } } })).map(i => [i.instanceId, i]))
    const instances = produced
      .filter(p => instMap.has(p.instanceId))
      .map(p => {
        const inst = instMap.get(p.instanceId) as any
        return { instanceId: p.instanceId, spaceId: p.spaceId, siteId: p.siteId, kind: inst.kind, status: inst.status, running: TwinEngine.owns(domainId, p.instanceId) }
      })
    return {
      source: r.source,
      connectorType: (r.scopeSpec as any)?.connectorType ?? r.adapterType,
      name: r.siteName,
      status: r.status,
      system: r.system,
      lastSyncedAt: r.lastSyncedAt,
      instanceCount: instances.length,
      instances
    }
  }

  /** 단건 상세(뷰/편집) — connectionConfig 포함, 어댑터 스키마의 secret 필드는 마스킹(있으면 '••••••'). */
  @Query(returns => ScalarObject, { nullable: true, description: 'Get one tenant-scoped reference with its connection config (secret fields masked) for view/edit.' })
  async twinReference(@Arg('source') source: string, @Ctx() context: ResolverContext): Promise<any | null> {
    const r = await getRepository(TwinReference).findOne({ where: { domain: { id: context.state.domain.id }, source } })
    if (!r) return null
    const secretKeys = new Set((getAdapter(r.adapterType)?.meta?.configSchema ?? []).filter(f => f.secret).map(f => f.key))
    const cfg: any = {}
    for (const [k, v] of Object.entries((r.connectionConfig as any) ?? {})) cfg[k] = secretKeys.has(k) ? (v ? '••••••' : '') : v
    /*
     * **이 연결이 만든 트윈 id 를 함께 낸다.** 상세 화면이 「이 연결이 잘 동기화되고 있나」를 물으려면
     * 어느 트윈을 물어야 하는지 알아야 한다. 예전에는 개수(`producedCount`)만 있어서 화면이 id 를
     * 알 방법이 없었다.
     */
    const producedInstanceIds = [
      ...new Set(
        ((r.scopeSpec as any)?.produced ?? [])
          .map((p: any) => p?.instanceId)
          .filter((x: any): x is string => typeof x === 'string' && !!x)
      )
    ]
    return {
      source: r.source, system: r.system, siteName: r.siteName, description: r.description,
      adapterType: r.adapterType, connectionConfig: cfg, status: r.status, lastSyncedAt: r.lastSyncedAt, lastError: r.lastError,
      producedInstanceIds,
      /* 목록과 같은 규칙 — 보낼 주소가 적힌 연결만 검증 상태를 싣는다(ADR-0046 ②). */
      ...(actuationAddressed(r.connectionConfig as any)
        ? {
            actuationState: r.actuationState ?? 'unverified',
            actuationVerifiedAt: r.actuationVerifiedAt ?? null,
            actuationFailure: r.actuationFailure ?? null
          }
        : {})
    }
  }

  @Query(returns => [ScalarObject], { description: 'List reference connector metadata (label, domain, config schema, availability, capabilities) for the connect UI.' })
  referenceAdapters(): any[] {
    /*
     * **능력을 함께 낸다** (2026-08-21).
     *
     * `capabilities` 는 어댑터 뿌리에 있고 `meta` 에는 없었다. 그래서 연결 화면이 「이 원본이 라이브를
     * 낼 수 있나」를 알 수 없었고, 트윈을 만들 때 `live` 를 말하지 못해 **재기동 정책이 없다고 거절**됐다
     * (실 연동에서 그것에 막혔다 — 미러는 정의상 `resync` 이므로 `live` 하나로 정해진다).
     *
     * 화면이 그 사실을 알아야 정책을 사람에게 물을지 스스로 정할지 가를 수 있다.
     */
    return listAdapters().map(a => ({ type: a.type, ...(a.meta ?? {}), capabilities: a.capabilities ?? [] }))
  }

  @Directive('@privilege(category: "twin", privilege: "mutation", domainOwnerGranted: true, superUserGranted: true)')
  @Mutation(returns => ScalarObject, { description: 'Create or update a tenant-scoped twin reference (persisted).' })
  async saveTwinReference(
    @Arg('source') source: string,
    @Arg('patch', type => ScalarObject) patch: any,
    @Ctx() context: ResolverContext
  ): Promise<any> {
    const domainId = context.state.domain.id
    const repo = getRepository(TwinReference)
    const existing = await repo.findOne({ where: { domain: { id: domainId }, source } })
    // 필드 단위 병합 — 편집 시 생략된 키(예: 마스킹된 secret)는 기존 값 유지.
    const connectionConfig = patch.connectionConfig
      ? { ...((existing?.connectionConfig as any) ?? {}), ...patch.connectionConfig }
      : existing?.connectionConfig

    /*
     * 같은 받는 사이트를 가리키는 연결 둘은 **저장하는 순간 거부한다** (ADR-0046 ②-1).
     * 들여오기(planImport)에는 있던 검사가 화면 저장에는 없어서, 연결을 베껴 이름만 바꾸면 두 트윈의
     * 조치가 한 공장으로 갔다 — 저쪽은 200 으로 받고 실제로 만든다. 짝(intakeEndpoint·intakeSiteId)은
     * 암호화된 칸 안에 있어 DB 유니크로 못 잡으니 여기서 잡는다(판정: connection-portability.ts).
     */
    const others = await repo.find({ where: { domain: { id: domainId } } })
    const taken = refuseIntakeCollision(source, connectionConfig as any, others as any)
    if (taken) return { ok: false, source, error: taken.message, errorCode: taken.code, errorParams: taken.params }

    /*
     * **주소를 바꾸는 저장은 검증을 되돌린다** (ADR-0046 ②). 지난 왕복은 지난 주소의 이야기라 새
     * 주소에 대해서는 아무 말도 하지 않는다. 화면이 그 초록을 그대로 두면 다른 공장을 가리키는 연결이
     * 「확인됨」으로 보인다. 되돌린 상태는 실패가 아니다 — 아직 안 해 본 것이다.
     */
    const reset = actuationResetPatch(existing?.connectionConfig as any, connectionConfig as any)

    const saved = await repo.save(
      repo.create({
        ...(existing ?? {}),
        ...(reset ?? {}),
        domain: { id: domainId } as any,
        source,
        system: patch.system ?? existing?.system,
        siteName: patch.siteName ?? existing?.siteName,
        description: patch.description ?? existing?.description,
        adapterType: patch.adapterType ?? existing?.adapterType,
        connectionConfig,
        mappingSpec: patch.mappingSpec ?? existing?.mappingSpec,
        scopeSpec: patch.scopeSpec ?? existing?.scopeSpec,
        status: patch.status ?? existing?.status ?? 'draft'
      })
    )
    return { ok: true, source: saved.source }
  }

  /** 연결 해제 — removeTwins=true 면 이 레퍼런스가 만든 트윈(provenance)도 함께 제거(라이프사이클 연쇄). 기본은 트윈 보존(분리). */
  @Directive('@privilege(category: "twin", privilege: "mutation", domainOwnerGranted: true, superUserGranted: true)')
  @Mutation(returns => Boolean, { description: 'Delete a tenant-scoped reference. If removeTwins, also remove the twin instances it produced (provenance cascade); otherwise twins are kept (detached).' })
  async deleteTwinReference(
    @Arg('source') source: string,
    @Ctx() context: ResolverContext,
    @Arg('removeTwins', { nullable: true }) removeTwins?: boolean
  ): Promise<boolean> {
    const domainId = context.state.domain.id
    const repo = getRepository(TwinReference)
    const ref = await repo.findOne({ where: { domain: { id: domainId }, source } })
    if (removeTwins && ref) {
      for (const p of (((ref.scopeSpec as any)?.produced ?? []) as any[])) {
        try {
          await TwinEngine.remove(domainId, p.instanceId)
        } catch {
          /* 이미 없거나 실패 — 연쇄는 best-effort. */
        }
      }
    }
    await repo.delete({ domain: { id: domainId } as any, source })
    return true
  }

  @Directive('@privilege(category: "twin", privilege: "mutation", domainOwnerGranted: true, superUserGranted: true)')
  @Mutation(returns => ScalarObject, { description: 'Test a reference connection via its adapter; updates status.' })
  async testReferenceConnection(@Arg('source') source: string, @Ctx() context: ResolverContext): Promise<any> {
    const repo = getRepository(TwinReference)
    const ref = await repo.findOne({ where: { domain: { id: context.state.domain.id }, source } })
    if (!ref) return { ok: false, error: `unknown reference: ${source}` }
    const adapter = getAdapter(ref.adapterType)
    if (!adapter) return { ok: false, error: `no adapter for type: ${ref.adapterType}` }
    try {
      const res = await adapter.testConnection(ref.connectionConfig ?? {})
      await repo.update({ id: ref.id }, { status: res.ok ? 'connected' : 'error', lastError: res.error ?? null } as any)
      return res
    } catch (e: any) {
      return { ok: false, error: e?.message ?? 'test failed' }
    }
  }

  /**
   * 어댑터 기반 인제스트 — discoverSites → 사이트별 fetchMaster → ingestMaster (fan-out, §2.2·§4.2).
   * P1 은 대개 N=1(virtual/단일). start 면 각 인스턴스 기동. 라이브 피드·mode 는 face2-inbound-live 소유.
   */
  /** 레퍼런스가 노출하는 사이트/공장 발견 — 트윈 생성 시 "어느 공장" 선택용(§2.2). */
  @Query(returns => ScalarObject, { description: 'Discover sites/factories exposed by a reference. Returns { sites, error? } — an empty list with no error means the system genuinely has none, which is different from a failure.' })
  async discoverReferenceSites(@Arg('source') source: string, @Ctx() context: ResolverContext): Promise<any> {
    /*
     * **"못 찾았다" 와 "없다" 를 구분한다.** 예전에는 넷을 빈 배열 하나로 합쳤고(레퍼런스 없음·어댑터
     * 미상·조회 실패·정말 0건), 화면은 그 모두를 "사이트를 찾지 못했습니다" 라고만 말했다. 실 시스템
     * 연결에서 가장 자주 막히는 자리인데 자격이 틀렸는지 주소가 틀렸는지 알 수 없었다.
     * 판정과 문장 형태는 `discovery-result`(순수)가 갖는다.
     */
    const ref = await getRepository(TwinReference).findOne({ where: { domain: { id: context.state.domain.id }, source } })
    if (!ref) return failed(referenceNotFound(source))
    const adapter = getAdapter(ref.adapterType)
    if (!adapter) return failed(adapterUnknown(ref.adapterType, source))
    try {
      /*
       * **기다려도 답이 없으면 사유를 준다.** 어댑터 계약에 시간 제한이 없어서, 주소가 틀린 엔드포인트는
       * 화면을 무한히 돌게 만들었다 — 무한히 도는 화면은 들어온 것이 없는 실패의 다른 얼굴이다.
       * 상한은 여기서 정한다(어댑터마다 다르게 두면 사용자가 어느 것을 기다리는지 알 수 없다).
       */
      const sites = await withTimeout(adapter.discoverSites(ref.connectionConfig ?? {}), DISCOVER_TIMEOUT_MS)
      return discovered(sites)
    } catch (e: any) {
      if (e?.code === TIMEOUT) return failed(discoveryTimedOut(source, DISCOVER_TIMEOUT_MS / 1000))
      return failed(discoveryFailed(source, detailOf(e)))
    }
  }

  /**
   * 트윈 생성 — 레퍼런스에서 사이트 발견 → (siteIds 있으면 그 공장만) → fetchMaster → provision. 사이트별 resilient.
   * 여러 공장을 관할하는 소스는 siteIds 로 "어느 공장"을 골라 생성(1:N, §2.1·§2.2).
   *
   * ── 현장 선택이 이 경로에도 있다 (2026-08-13) ────────────────────────────
   * 한 현실을 여러 렌즈가 비춘다(WMS·MES·YMS). 그런데 **그 선택이 템플릿 경로에만** 있어서
   * (`createFromTemplate(spaceId)`), 실 시스템에서 가져온 트윈은 어댑터가 파생한 공간 id 가 우연히
   * 일치할 때만 한 현장에 모였다 — 사람이 "이 MES 와 같은 현장에 넣어라" 를 말할 방법이 없었다.
   * 데이터로는 가능한 일이 저작으로는 불가능한, `ingest-space` 가 이미 지적한 그 비대칭이다.
   *
   * 판정은 그 순수 모듈이 갖는다: 없는 공간은 거절(`resolveIngestSpace`), 사이트 여럿을 한 공간에
   * 넣는 것도 거절(`refuseImportSpace`) — 서로 다른 현장이 알리지 않고 합쳐지는 것을 막는다.
   */
  @Directive('@privilege(category: "twin", privilege: "mutation", domainOwnerGranted: true, superUserGranted: true)')
  @Mutation(returns => ScalarObject, {
    description:
      "Create twins from a reference: discover sites → filter by siteIds (if given) → fetch master(s) → provision. spaceId co-locates the twin into an existing space (one site only). start=false to only provision; live=true to run as a mirror (adapter openLiveFeed → live projector) instead of sim. restartPolicy declares how each new twin treats its own past on reboot ('resync' | 'resume' | 'reset'); omit it and live=true implies 'resync' — a mirror reads the source again — while a simulation must declare 'resume' or 'reset' for itself."
  })
  async importTwinReference(
    @Arg('source') source: string,
    @Ctx() context: ResolverContext,
    @Arg('siteIds', type => [String], { nullable: true }) siteIds?: string[],
    @Arg('start', { nullable: true }) start?: boolean,
    @Arg('live', { nullable: true }) live?: boolean,
    @Arg('spaceId', { nullable: true, description: 'Existing space to co-locate this twin into (add a lens to a site). Only when importing a single site. Omit to give each site its own space.' }) spaceId?: string,
    @Arg('restartPolicy', { nullable: true, description: "How each new twin treats its own past on reboot: 'resync' | 'resume' | 'reset'. Required unless live=true (a mirror is 'resync' by definition)." }) restartPolicy?: string
  ): Promise<any> {
    const domainId = context.state.domain.id
    const repo = getRepository(TwinReference)
    const ref = await repo.findOne({ where: { domain: { id: domainId }, source } })
    if (!ref) return { ok: false, error: `unknown reference: ${source}` }
    const adapter = getAdapter(ref.adapterType)
    if (!adapter) return { ok: false, error: `no adapter for type: ${ref.adapterType}` }
    const cfg = ref.connectionConfig ?? {}
    try {
      let sites = await adapter.discoverSites(cfg)
      if (siteIds?.length) sites = sites.filter(s => siteIds.includes(s.siteId))
      /* 아무것도 쓰기 전에 거절한다 — 한 사이트를 넘으면 "이 현장에 더한다" 가 성립하지 않는다. */
      const refusal = refuseImportSpace(sites.length, spaceId)
      if (refusal) return { ok: false, error: refusal.message, errorCode: refusal.code, errorParams: refusal.params }
      const instances: any[] = []
      const failed: any[] = []
      /*
       * **진행을 구독으로 흘린다** (2026-08-21).
       *
       * 마스터 읽기는 원본에 여러 질의를 보내는 긴 작업이고, 화면은 그 사이 아무것도 알 수 없었다.
       * 단계를 아는 것은 어댑터뿐이므로 어댑터가 말하고(표준 낱말), 호스트가 자기 두 단계(생성·기동)를
       * 더해 흘린다.
       *
       * **분모는 어댑터가 선언한 것**이다(`masterSteps`). 선언하지 않은 어댑터는 `total` 이 없고, 그때
       * 화면은 퍼센트를 만들지 않는다 — 모르는 진행률을 지어내지 않는다.
       */
      const declared = adapter.masterSteps?.length
      const stepTotal = declared === undefined ? undefined : declared + HOST_STEPS.length
      const emit = (p: Omit<ReferenceProgress, 'siteCount'>) =>
        pubsub.publish(REFERENCE_PROGRESS_TOPIC, referenceProgressPayload(domainId, { ...p, siteCount: sites.length, total: p.total ?? stepTotal, source }))

      /*
       * **읽은 뒤의 총평을 사이트마다 남긴다** (2026-08-22).
       *
       * 총평은 읽는 순간에만 있었다 — 대화상자를 닫으면 사라지고, 다시 보려면 같은 읽기를 또 해야 했다.
       * 그런데 사람이 그것을 묻는 자리는 「모델 살펴보기」다(트윈을 보다가 「이건 왜 비었나」를 묻는다).
       * 그래서 레퍼런스 행에 남겨 그 화면이 언제든 읽게 한다.
       *
       * 사실만 담는다(`assessMaster`) — 문장은 화면이 만들고, 더 알고 싶을 때 AI 가 이 사실만 근거로
       * 서술한다. 다시 읽으면 그 사이트의 총평이 새것으로 바뀐다(지난 읽기의 총평을 현재로 보이게
       * 두지 않는다).
       */
      const assessments: Record<string, unknown> = {}

      for (const [siteIndex, site] of sites.entries()) {
        /*
         * **실패를 실제로 막힌 단계에 적는다** (2026-08-22).
         *
         * 예전에는 실패를 무조건 `provision` 으로 적었다. 그래서 원본을 읽다 끊긴 것도 화면에는
         * 「트윈 생성에서 실패」로 떴다 — 사람이 원인을 엉뚱한 자리에서 찾는다(실제로 그렇게 됐다).
         *
         * 규약이 답을 준다: `done` 은 **끝낸 단계 수**다(`ReferenceProgress.done`). 그러면 지금 막힌 것은
         * 그 다음 단계, 즉 `masterSteps[done]` 이다. 그 표를 넘어서면(호스트 단계) 마지막으로 알린 것을 쓴다.
         *
         * 이 규칙은 두 가지 보고 방식에 함께 맞는다 — 단계 **시작**에 알리는 어댑터(`done` 이 아직 0)와
         * 단계 **완료**에 알리는 어댑터(`done` 이 1 늘어난다) 모두 「끝낸 수」만 지키면 된다.
         */
        let lastDone = 0
        let lastStep: ReferenceProgress['step'] = (adapter.masterSteps?.[0] ?? 'provision') as ReferenceProgress['step']
        const track = (p: Omit<ReferenceProgress, 'siteCount'>) => {
          lastDone = p.done
          lastStep = p.step
          emit(p)
        }
        try {
          const master = await adapter.fetchMaster(cfg, site, st =>
            track({ siteId: site.siteId, siteIndex, step: st.key, done: st.done, total: stepTotal, detail: st.detail })
          )
          /* 어느 레퍼런스의 **어느 사이트**에서 왔는지 남긴다 — 다시 읽기가 같은 어댑터·같은 사이트로 돌아오려면
             사이트 정체가 있어야 한다(`source` 는 참조 이름이라 사이트 전부에 같다). */
          /*
           * 정책은 **선언이거나 의도에서 곧바로 따라오는 것**이다: `live=true` 는 미러를 띄우겠다는 뜻이고
           * 미러는 정의상 `resync` 다(외부가 진실이므로 우리 과거를 이어 붙이지 않는다). 시뮬로 띄우려면
           * 스스로 선언해야 한다 — 그 자리에서 우리가 고르면 재기동마다 이력을 지우는 트윈이 알리지 않고 생긴다.
           */
          assessments[site.siteId] = assessMaster(master, site.siteId)
          const policy = readRestartPolicy(restartPolicy ?? (live ? 'resync' : undefined), `importTwinReference("${source}")`)
          track({ siteId: site.siteId, siteIndex, step: 'provision', done: declared ?? 0, total: stepTotal })
          const res = await TwinEngine.ingestMaster(
            domainId,
            { ...master, origin: { kind: 'reference', source, siteId: site.siteId, ...(spaceId ? { spaceId } : {}) } },
            policy,
            { spaceId },
            context.state.user
          )
          track({ siteId: site.siteId, siteIndex, step: 'start', done: (declared ?? 0) + 1, total: stepTotal })
          if (start !== false) {
            // live=true + 어댑터가 openLiveFeed 지원 → mirror(실 이벤트 미러). 아니면 sim 기동(fallback).
            /* 읽기 커서를 저장할 자리를 함께 넘긴다 — 레퍼런스의 이름은 호스트가 붙이는 것이고
               커넥터는 그것을 알 수 없다(§`startReferenceLiveFeed`). */
            const wentLive = live
              ? await startReferenceLiveFeed(domainId, ref.adapterType, cfg, site, res.instanceId, ref.source)
              : false
            if (!wentLive) await TwinEngine.startFromRegistry(domainId, res.instanceId)
            instances.push({ siteId: site.siteId, instanceId: res.instanceId, spaceId: res.spaceId, mode: wentLive ? 'live' : 'sim', warnings: res.warnings })
          } else {
            instances.push({ siteId: site.siteId, instanceId: res.instanceId, spaceId: res.spaceId, mode: 'provisioned', warnings: res.warnings })
          }
        } catch (e: any) {
          /*
           * **실패한 단계를 그대로 알린다.** 화면이 진행 표시를 지우지 않고 그 자리를 붉게 두어야
           * 사람이 「어디까지 갔는지」를 안다. 막힌 자리는 「끝낸 수」의 다음 단계다(위 주석).
           */
          const at = (adapter.masterSteps?.[lastDone] ?? lastStep) as ReferenceProgress['step']
          emit({ siteId: site.siteId, siteIndex, step: at, done: lastDone, total: stepTotal, failed: true, error: e?.message ?? String(e) })

          failed.push({ siteId: site.siteId, error: e?.message ?? 'failed', ...(e?.code ? { errorCode: e.code, errorParams: e.params } : {}) })
        }
      }
      // provenance — 이 레퍼런스가 만든 인스턴스 추적(scopeSpec.produced). 삭제 연쇄·상태 동기·네비게이션 근거.
      const producedNew = instances.map(i => ({ siteId: i.siteId, instanceId: i.instanceId, spaceId: i.spaceId }))
      const prev: any[] = ((ref.scopeSpec as any)?.produced ?? []).filter((p: any) => !producedNew.some(n => n.instanceId === p.instanceId))
      await repo.update(
        { id: ref.id },
        {
          status: 'connected',
          lastSyncedAt: new Date().toISOString(),
          /* 못 옮긴 값을 **세어 남긴다**. 아무 인스턴스도 보고하지 않았으면 `null` — 「없었다」로 위장하지 않는다. */
          lastWarningCount: instances.some((i: any) => Array.isArray(i.warnings))
            ? instances.reduce((n: number, i: any) => n + (Array.isArray(i.warnings) ? i.warnings.length : 0), 0)
            : null,
          /*
           * **영어 canonical 로 남긴다.** 이 값은 DB 에 저장되고 연결 화면이 그대로 뿌린다 — 예전에는
           * 한글 산문(`N개 사이트 실패`)이라 다섯 언어 화면에 한글이 새어 나갔다. 어댑터가 주는 사유도
           * 같은 필드에 들어오므로(그것은 원 시스템의 말이다) 키가 아니라 영어 문장으로 통일한다.
           * 어느 사이트가 왜 실패했는지는 `failed[]` 가 답한다 — 화면이 그것을 보여 준다.
           */
          lastError: failed.length ? `${failed.length} of ${sites.length} site(s) failed: ${failed.map((f: any) => f.siteId).join(', ')}` : null,
          /*
           * ── 실패 사유를 **남긴다** (2026-08-21) ────────────────────────────────
           *
           * 사유는 대화상자에만 있었다. 닫으면 사라지고, 다시 보려면 같은 실패를 다시 만들어야 했다 —
           * 실제로 「승화푸드」 연결이 그 상태였다(요약 한 줄만 남고 원인은 어디에도 없었다).
           *
           * `scopeSpec` 은 이미 이 레퍼런스의 부속 사실을 담는 자리이므로(`produced`) 스키마를 늘리지 않고
           * 여기 적는다. 성공한 회차에는 **비운다** — 지나간 실패를 현재 상태처럼 보이게 두지 않는다.
           */
          scopeSpec: {
            ...((ref.scopeSpec as any) ?? {}),
            produced: [...prev, ...producedNew],
            /* 이번에 읽은 사이트의 총평만 갈아 끼운다 — 읽지 않은 사이트의 것은 그대로 둔다. */
            assessments: { ...(((ref.scopeSpec as any) ?? {}).assessments ?? {}), ...assessments },
            ...(failed.length
              ? { lastFailures: failed.map((f: any) => ({ siteId: f.siteId, error: f.error, at: new Date().toISOString() })) }
              : { lastFailures: [] })
          }
        } as any
      )
      return { ok: instances.length > 0, instances, failed }
    } catch (e: any) {
      /* 레퍼런스 행에는 **영어 canonical** 을 남긴다 — 저장되는 값은 언어중립이어야 다시 읽을 수 있다. */
      await repo.update({ id: ref.id }, { status: 'error', lastError: e?.message ?? 'ingest failed' } as any)
      return refusal(context, e, 'ingest failed')
    }
  }
}