import { twinLog, twinWarn, twinError } from '../../engine/log.js'
import { getRepository } from '@things-factory/shell'

import { getAdapter, type ConnectionConfig, type LiveFeedContinuity, type ObservedItemFact, type SiteDescriptor } from './reference-adapter.js'
/* 이름 짓는 규칙은 가벼운 파일에 있다 — 테스트가 붙게(§`live-feed-lease.ts`). */
import { liveFeedLeaseName } from './live-feed-lease.js'
export { liveFeedLeaseName } from './live-feed-lease.js'
import { databaseCommandStore } from '../actuation/command-store.js'
import { settleEffects } from '../actuation/settle-effects.js'
import { TwinEngine, ingestCanonicalRecords, registerLiveFeedProbe } from '../../engine/index.js'
import { type CanonicalRecord } from '@operato/ops-contract'
import { TwinInstance } from '../twin-instance/twin-instance.js'
import { TwinReference } from './twin-reference.js'

/*
 * 커넥터 기반 라이브 피드 결선(reference → live, §0 프레임 ① restartPolicy='resync').
 * 흐름: adapter.openLiveFeed(cfg, site, records) → ingestCanonicalRecords(정규→봉투) → TwinEngine.ingestLive(projector 미러).
 * 커넥터가 "벤더 이벤트 → 정규 레코드" 매핑을 소유(코드)한다 — preset 기반 startLiveFeed(operato-twin, http-get+jsonata)와
 * 대칭이되, 매핑 위치만 다르다. 실 시스템 전환은 connectionConfig.baseUrl 교체뿐(코드 경로 동일).
 * 생명주기: 인스턴스 정지(TwinEngine.stop)에 훅으로 묶여 피드도 함께 unsubscribe.
 */

/** 활성 라이브 피드 — instanceId → unsubscribe(어댑터 openLiveFeed 반환). */
const feeds = new Map<string, () => void>()

/**
 * 커넥터 라이브 기동 — 어댑터 openLiveFeed 를 인스턴스에 결선.
 * 반환: 라이브 결선 성공 여부(어댑터가 openLiveFeed 미지원이면 false → 호출측이 sim 폴백).
 */
export async function startReferenceLiveFeed(
  domainId: string,
  adapterType: string,
  cfg: ConnectionConfig,
  site: SiteDescriptor,
  instanceId: string,
  /**
   * **어느 레퍼런스에서 읽는 피드인가** — 읽기 커서를 저장할 자리를 가리킨다(§`loadLiveCursor`).
   *
   * 부르는 쪽이 `TwinReference` 를 이미 들고 있으므로 그 이름을 넘긴다. 처음에는 `cfg`·`site` 에서
   * 찾으려 했는데 **둘 다 그 값을 가질 수 없다**: `cfg` 는 사람이 화면에서 입력한 접속 설정이고
   * (`endpoint`·`token`…), `site` 는 커넥터가 원본에서 발견한 것이다 — **레퍼런스의 이름은 호스트가
   * 붙이는 것이고 원본은 그것을 모른다.** 커넥터가 채울 수 없는 값을 커넥터에서 읽으려 한 것이 잘못이었다.
   *
   * 주지 않으면 커서를 잇지 않고 **그 사실을 말한다**(알리지 않고 창이 미끄러지지 않게).
   */
  source?: string
): Promise<boolean> {
  const adapter = getAdapter(adapterType)
  if (!adapter?.openLiveFeed) return false

  const det = await TwinEngine.detail(domainId, instanceId)
  if (!det?.model) throw new Error(`instance "${instanceId}" not provisioned (no model) — ingest master before live start`)

  // 관측 기동 — 커널 tick 없이 projector 가 실 이벤트를 반영한다(startLive 가 restartPolicy='resync' 로 적는다).
  // 시각 기준은 **공간**이 갖는다(테넌트 아님) — 교대의 HH:MM 을 어느 기준으로 읽나.
  TwinEngine.startLive(instanceId, domainId, det.kind, await TwinEngine.withSpaceTimeBase(det.model, domainId))

  stopReferenceLiveFeed(instanceId) // 재기동 시 이전 피드 정리
  /*
   * **어디까지 읽었나를 이 층이 들고 건넨다**(§`LiveFeedContinuity`, 2026-08-23).
   *
   * 어댑터가 이 값을 자기 사물함에 두면 원본이 늘 때마다 저장 기제가 늘고, 그중 하나가 알리지 않고 다르게
   * 동작한다. 그리고 커널이 이미 갈라 두었다: 「원천이 애초에 다시 말해 주지 않는 축」은 재기동
   * 연속성으로 이어받는다(§`hydrateContinuity`) — 읽기 커서가 정확히 그 성질이다.
   *
   * The object itself is handed over even on a first attach, because the feed needs its lease name
   * before it has read anything. `cursor` staying absent is what says "first attach" — which is how
   * every connector already reads it (`continuity?.cursor`), so nothing changes for them. What is
   * never filled in with an empty value is the cursor: an empty one would make a first attach look
   * like a resumed one, and the adapter would build no look-back window.
   */
  const carried = await loadLiveCursor(domainId, instanceId, source)
  const continuity: LiveFeedContinuity = { ...carried, leaseName: liveFeedLeaseName(domainId, instanceId, source) }
  /*
   * The lease name is composed here, not in the adapter, because it has to be unique across
   * domains and an adapter is handed neither `domainId` nor `instanceId` (§`LiveFeedContinuity`).
   * Built from `source` rather than `site.siteId`: one reference is one connection to one system,
   * and that is the thing a second instance must not open twice.
   */
  /*
   * ── **트윈이 아는 것을 함께 건넨다** (2026-08-28) ───────────────────────────
   *
   * 전량 절대값을 읽는 커넥터는 지난 주기와 견주어 바뀐 것만 낸다. 그 비교 표가 프로세스 안에만
   * 있어서 재기동하면 커넥터가 전량을 「바뀐 것」으로 판정했다 — 승화푸드에서 2,665건이다.
   *
   * 트윈은 저장본에서 되세우므로 잃지 않았다. 그래서 그 사실을 건네면 첫 주기부터 바뀐 것만 나간다.
   * 커널 어휘로 건넨다 — 원본 낱말로 되돌리는 일은 여러 낱말이 한 값으로 모여 불가능하고, 필요도 없다.
   */
  continuity.observedItems = observedItemsOf(domainId, instanceId)
  const unsubscribe = adapter.openLiveFeed(cfg, site, (records: unknown[]) => {
    if (!TwinEngine.owns(domainId, instanceId)) {
      stopReferenceLiveFeed(instanceId) // 인스턴스 사라지면 자가 정지(안전망)
      return
    }
    const offered = Array.isArray(records) ? records.length : records ? 1 : 0
    const { accepted, rejected, masterData } = ingestCanonicalRecords(
      records as CanonicalRecord[],
      domainId,
      new Date().toISOString(),
      TwinEngine.factScope(domainId, instanceId)
    )
    /*
     * ── **사건이 아닌 것은 상태만 세운다** (2026-08-28) ─────────────────────────
     *
     * 마스터데이터 — 변하지 않는 속성(유통기한·로트번호). 시각축이 없는 값이므로 지난 기록에 적지
     * 않는다(§`applyMasterData`). 이 문이 없던 동안 그 값은 사건에 실려 왔고, 표준이 그것을
     * 「들어왔다」에만 허용하므로 전량을 읽는 원본은 **들어오지 않은 것을 들어왔다고** 말해야 했다.
     *
     * 반영된 건수를 함께 센다 — 라이브로 돌지 않는 트윈은 0 을 내므로, 그 차이가 「바닥에 버려진 수」다.
     */
    const masterApplied = masterData.length ? TwinEngine.applyMasterData(domainId, instanceId, masterData) : 0
    if (masterData.length && masterApplied < masterData.length) {
      twinWarn(
        `[twin-live] "${instanceId}": ${masterData.length - masterApplied}/${masterData.length} master data element(s) NOT applied — ` +
          'the twin is not running live; lot attributes such as expiry will stay empty.'
      )
    }
    /*
     * 커널이 **몇 건을 실제로 받았나**를 함께 본다. 트윈이 라이브로 돌지 않으면 `ingestLive` 가 0 을
     * 돌려주고, 그 차이가 곧 「바닥에 버려진 수」다. 이것을 통과로 세면 멈춘 트윈이 100% 로 보인다.
     */
    /* 어떤 길로 들어왔는지 함께 적는다 — 화면이 「이 트윈은 무엇을 통해 받나」를 말할 수 있게. */
    const applied = accepted.length ? TwinEngine.ingestLive(domainId, instanceId, accepted, 'connector') : 0

    /*
     * **나간 지시와 잇는다** — 폐루프. 훅과 같은 자리다(§`routes.ts`). 기다리지 않는다: 라이브 묶음
     * 처리가 이것 때문에 늦으면 뒤의 묶음이 밀리고, 밀린 것이 크래시로 이어진 적이 있다.
     */
    void settleEffects({ domainId, envelopes: accepted, store: databaseCommandStore() })
    const undelivered = accepted.length - applied
    /*
     * **거부한 것을 알리지 않고 버리지 않는다.** 예전에는 `rejected` 를 받지도 않았다 — 커넥터의 매핑이
     * 틀리면 트윈은 「아무 일도 일어나지 않는」 모습이 되고, 로그에도 아무 흔적이 없었다.
     *
     * 그다음 자리였던 「화면까지 나르기」를 이제 채운다. 건마다 브로드캐스팅하지 않고 **집계**한다 — 통과분까지
     * 함께 세는 이유는, 거부만 세면 「유입 없음」과 「정상」이 둘 다 거부 0 으로 같아 보이기 때문이다.
     * 조회는 `twinIngestHealth`.
     */
    /*
     * **읽었다는 것 자체가 사실이다** — 빈 읽기도 성공이므로 단절 기록을 지운다(§`onReadFailure`).
     * 그 둘을 섞으면 들어온 것이 없는 연결이 끊긴 원본으로 보인다.
     */
    TwinEngine.clearIngestReadFailure(domainId, instanceId)
    TwinEngine.recordIngestResult(domainId, instanceId, offered, rejected, Date.now(), undelivered)
    if (rejected.length) {
      const why = rejected[0]?.errors?.join(' · ') ?? 'unknown'
      twinWarn(`[twin-live] "${instanceId}": ${rejected.length}/${offered} record(s) rejected — e.g. ${why}`)
    }
    /* 버려진 것은 로그에도 남긴다 — 화면을 보지 않는 사람도 알아야 한다(사실이 사라지는 중이다). */
    if (undelivered > 0) {
      twinWarn(
        `[twin-live] "${instanceId}": ${undelivered}/${offered} record(s) NOT delivered — the twin is not running live; ` +
          `records are being discarded. Start the twin, or disconnect the feed.`
      )
    }
  }, continuity)
  feeds.set(instanceId, unsubscribe)
  return true
}

/**
 * 트윈이 지금 들고 있는 물품을 커널 어휘로 낸다 — 없으면 `undefined`.
 *
 * 상태가 없으면(트윈이 아직 아무것도 모르면) `undefined` 다. 빈 배열로 메우지 않는다 — 「모른다」와
 * 「하나도 없다」는 다른 사실이고, 커넥터가 그 둘에 다르게 반응해야 한다.
 */
function observedItemsOf(domainId: string, instanceId: string): ObservedItemFact[] | undefined {
  const state = TwinEngine.snapshot(domainId, instanceId)
  const items = (state as any)?.items
  if (!Array.isArray(items)) return undefined
  return items.map((i: any) => ({
    id: String(i?.subLotId ?? i?.epc ?? ''),
    ...(i?.gtin ? { epcClass: String(i.gtin) } : {}),
    ...(i?.location ? { location: String(i.location) } : {}),
    ...(typeof i?.qty === 'number' ? { qty: i.qty } : {}),
    ...(i?.uom ? { uom: String(i.uom) } : {}),
    ...(i?.disposition ? { disposition: String(i.disposition) } : {}),
    ...(typeof i?.expiry === 'number' ? { expiry: i.expiry } : {}),
    ...(i?.lot ? { lot: String(i.lot) } : {})
  })).filter((i: ObservedItemFact) => !!i.id)
}

/**
 * 그 원본의 읽기 커서를 불러오고, 움직이면 저장하는 통로를 만든다.
 *
 * 커서는 **원본(`TwinReference`)의 것**이지 인스턴스의 것이 아니다 — 한 원본이 여러 인스턴스를 낳고
 * (`scopeSpec.produced`) 읽는 것은 한 번이다. 인스턴스마다 커서를 두면 같은 표를 여러 번 읽는다.
 *
 * 저장 실패는 **삼키지 않고 말한다.** 오류 없이 실패하면 다음 재기동에서 창이 다시 미끄러지고, 그 손실은
 * 오류 없이 사라진다 — 이 컬럼을 만든 이유가 바로 그것이다.
 */
async function loadLiveCursor(
  domainId: string,
  /** 읽기 실패를 적을 트윈 — 커서는 레퍼런스의 것이지만 **건강은 인스턴스의 것**이다(한 레퍼런스가 여럿을 낳는다). */
  instanceId: string,
  source?: string
  /*
   * `leaseName` is not this function's to know — it is composed from the domain and instance by the
   * caller. Returning the rest keeps that rule in one place instead of two.
   */
): Promise<Omit<LiveFeedContinuity, 'leaseName'> | undefined> {
  if (!source) {
    /* 원본을 가리킬 수 없으면 저장할 자리도 없다 — 그 사실을 말한다(알리지 않고 커서 없이 돌지 않게). */
    twinWarn('[twin-live] live cursor not carried — the caller did not say which reference this feed reads from')
    return undefined
  }
  const repo = getRepository(TwinReference)
  const row = await repo.findOne({ where: { domain: { id: domainId } as any, source } }).catch(() => null)
  if (!row) {
    twinWarn(`[twin-live] live cursor not carried — no reference row for source "${source}"`)
    return undefined
  }
  const stored = row.liveCursor
  if (stored?.firstAttachedAt) {
    twinLog(`[twin-live] "${source}": resuming from stored cursor — knowledge starts ${stored.firstAttachedAt}`)
  }
  return {
    cursor: stored ?? undefined,
    /*
     * **원본에 닿지 못했다** — 어댑터가 알리면 상태에 적는다(§`onReadFailure`).
     *
     * 로그로는 어댑터가 이미 말하고 있었다(재시도 경고). 그러나 **로그는 사람이 볼 때만 값이 있다** —
     * 화면이 말하려면 조회 가능한 상태에 있어야 한다.
     *
     * ── `status` 를 건드리지 않는다 (2026-08-23, 되돌린 자리) ────────────────────
     * 처음에는 레퍼런스 행의 `status` 를 `'error'` 로도 적었다 — 「목록이 「연결됨」이라 말하면서 실제로는
     * 못 읽는 상태를 없애려고」. **그것이 피드를 영구히 껐다.**
     *
     * 부팅 재부착은 `status: 'connected'` 인 레퍼런스만 되살린다(§`resumeReferenceLiveFeeds`). 그래서
     * 원본이 **잠깐** 끊긴 것이 그 행을 `error` 로 만들고, 다음 재기동에서 그 피드는 아예 붙지 않았다 —
     * 실측으로 그렇게 났다(09:25 버전에 `[twin-live]` 줄이 한 줄도 없었다).
     *
     * 두 축을 한 칸에 섞은 것이 원인이다:
     *   `status`      **연결 설정이 성립하나** — 접속 시험·마스터 동기가 정하는 축(재부착의 자격이다)
     *   읽기 실패      **지금 닿나** — 오갈 수 있는 사실이고, 유입 장부가 그 축을 갖는다
     *
     * 그러니 여기서는 `lastError` 에만 적는다(사람이 목록에서 이유를 볼 수 있게). 「지금 닿나」는
     * `ingestHealth().readFailure` 가 답하고, 성공한 주기가 그것을 지운다.
     */
    onReadFailure: ({ reason, stream, nextRetryMs }) => {
      TwinEngine.recordIngestReadFailure(domainId, instanceId, reason, Date.now(), stream, nextRetryMs)
      twinWarn(`[twin-live] "${source}": cannot read from the source${stream ? ` (${stream})` : ''} — ${reason}`)
      repo
        .update({ id: row.id }, { lastError: `live: ${reason}` } as any)
        .catch(err => twinWarn(`[twin-live] "${source}": read failure not recorded — ${err?.message ?? err}`))
    },
    /*
     * **끊겼다가 돌아왔다** — 그 사실을 장부에 남긴다 (2026-08-27).
     *
     * 실패 기록을 지우기만 하면 「끊긴 적이 있었다」가 화면에서 사라진다. 그러면 밤새 두 시간 끊겼던
     * 연결과 한 번도 끊기지 않은 연결이 아침에 똑같이 보인다.
     *
     * `lastError` 도 함께 지운다 — 목록에 남은 옛 이유가 지금의 상태로 읽힌다.
     */
    onRecovered: ({ afterFailures, downMs }) => {
      TwinEngine.recordIngestRecovered(domainId, instanceId, afterFailures, downMs, Date.now())
      twinLog(
        `[twin-live] "${source}": reading again after ${afterFailures} failed attempt(s) — it was out for ${Math.round(downMs / 1000)}s`
      )
      repo
        .update({ id: row.id }, { lastError: null } as any)
        .catch(err => twinWarn(`[twin-live] "${source}": recovery not recorded — ${err?.message ?? err}`))
    },
    /**
     * **읽었는데 창을 넘길 수 없다** — 위와 조치가 반대인 사실이다 (2026-08-24).
     *
     * 이 문을 만들기 전에는 이 사실이 `onReadFailure` 로 나갔다. 그래서 화면이 「원본에 닿지 못한다」고
     * 말했는데 실제로는 원본이 **답한** 상태였고, 그 답의 모양이 커서를 이긴 것이었다. 사람은 원본을
     * 의심하고 기다리는데 기다림으로는 영원히 풀리지 않는다 — **원인을 반대 방향으로 가리켰다.**
     *
     * `status` 를 건드리지 않는 규율은 위와 같다(재부착의 자격이다). `lastError` 에는 적는다 — 목록에서
     * 이유를 볼 수 있어야 하고, 「닿지 못한다」와 다른 문장이어야 한다.
     */
    /*
     * **세우지 않은 것을 장부에 적는다** — 원본에 있는데 트윈에 세우지 않은 것(이유와 수).
     *
     * 로그로도 남기지만 **로그는 사람이 볼 때만 값이 있다.** 이 사실은 상태의 성질이라(세우지 않은 것은
     * 지금도 없다) 조회되는 자리에 있어야 한다 — 그러지 않으면 프로비저닝 화면에서 한 번 스쳐 지나간다.
     *
     * `status`·`lastError` 를 건드리지 않는다: 이것은 **실패가 아니다.** 어댑터가 옳은 판단으로 안 받은
     * 것이고, 오류로 적으면 화면이 원본을 의심하게 만든다.
     */
    onWithheld: ({ reason, count }) => {
      TwinEngine.recordIngestWithheld(domainId, instanceId, reason, count)
    },
    onCursorStall: ({ reason, stream }) => {
      TwinEngine.recordIngestCursorStall(domainId, instanceId, reason, Date.now(), stream)
      twinWarn(
        `[twin-live] "${source}": read the source but the cursor cannot advance` +
          `${stream ? ` (${stream})` : ''} — ${reason}. Waiting will not clear this.`
      )
      repo
        .update({ id: row.id }, { lastError: `live cursor stalled: ${reason}` } as any)
        .catch(err => twinWarn(`[twin-live] "${source}": cursor stall not recorded — ${err?.message ?? err}`))
    },
    onCursor: next => {
      /* 어댑터가 준 모양을 그대로 적는다 — 이 층은 흐름의 뜻을 모른다(열쇠는 어댑터가 정한다). */
      repo
        .update({ id: row.id }, { liveCursor: next } as any)
        .catch(err => twinWarn(`[twin-live] "${source}": live cursor not saved — ${err?.message ?? err}`))
    }
  }
}

/** 라이브 피드 중지(unsubscribe). */
export function stopReferenceLiveFeed(instanceId: string): boolean {
  const unsub = feeds.get(instanceId)
  if (!unsub) return false
  try { unsub() } catch { /* noop */ }
  feeds.delete(instanceId)
  return true
}

/** 활성 피드 목록(관측). */
export function listReferenceLiveFeeds(): string[] {
  return [...feeds.keys()]
}

/**
 * 부팅 때 **끊어진 피드를 다시 붙인다** — 도는 미러 트윈에 계측이 계속 들어오게.
 *
 * ── 무엇이 알리지 않고 끊겼나 (2026-08-14) ───────────────────────────────────────
 * `startReferenceLiveFeed` 를 부르는 곳은 `importTwinReference` **하나뿐**이었다. 그래서 서버를 한 번
 * 재기동하면 미러 트윈은 「도는 중」이라 말하면서 아무 계측도 받지 않았다 — 오류도 로그도 없이,
 * 사람은 「값이 안 변하네」로 알게 된다. 실 배포에서 재기동 한 번에 계측이 멈추는 셈이다.
 *
 * ── 되살릴 재료는 이미 영속돼 있다 ─────────────────────────────────────────
 * `TwinReference` 가 어댑터 종류·접속 설정·그 레퍼런스가 만든 인스턴스(`scopeSpec.produced`)를 들고 있다.
 * 사이트 서술자만 없는데(`hint` 는 남기지 않았다) **다시 찾아서** 얻는다(`discoverSites`) — 인제스트가
 * 걷는 것과 같은 길이므로 규칙이 갈라지지 않는다. 되살릴 인스턴스가 없으면 그 호출도 하지 않는다.
 *
 * 재개 대상은 **등록부가 도는 미러라고 말하는 것**뿐이다: 멈춘 트윈을 되살리면 사람이 멈춘 것을
 * 우리가 다시 켜는 것이고, 시뮬 트윈에 실 계측을 부으면 두 현실이 섞인다.
 */
export async function resumeReferenceLiveFeeds(): Promise<{ resumed: string[]; failed: { instanceId: string; error: string }[] }> {
  const resumed: string[] = []
  const failed: { instanceId: string; error: string }[] = []
  /*
   * **한 트윈에 피드는 하나다** (2026-08-18).
   *
   * 같은 사이트를 두 레퍼런스가 각각 인제스트하면 둘 다 그 인스턴스를 자기가 만든 것(`produced`)으로
   * 들고 있다. 기동 때 그 목록을 그대로 돌면 같은 트윈에 피드를 두 번 붙이는데, 붙이는 함수가 미러를
   * 다시 세우므로(`startLive`) **먼저 쌓인 수요 구간이 리셋된다** — 실제로 그렇게 났다
   * (로그: `reattached 2 feed(s): ems-plant-1, ems-plant-1`).
   *
   * 그러니 한 번만 붙이고, **두 레퍼런스가 같은 트윈을 주장한다는 사실은 말한다** — 알리지 않고 하나를
   * 고르면 어느 접속이 그 트윈을 먹이고 있는지 아무도 모르게 된다.
   */
  const attached = new Map<string, string>() // instanceId → 먼저 붙인 레퍼런스
  try {
    const refs = await getRepository(TwinReference).find({ where: { status: 'connected' }, relations: ['domain'] })
    for (const ref of refs) {
      const domainId = (ref as any).domain?.id ?? (ref as any).domainId
      const adapter = getAdapter(ref.adapterType)
      if (!domainId || !adapter?.openLiveFeed) continue // 관측만 하는 어댑터에는 되살릴 피드가 없다

      const produced: any[] = ((ref.scopeSpec as any)?.produced ?? []).filter((p: any) => p?.instanceId && p?.siteId)
      if (!produced.length) continue

      /* 등록부가 「도는 미러」라고 말하는 것만 고른다. */
      const rows = await getRepository(TwinInstance).find({
        where: produced.map((p: any) => ({ domain: { id: domainId }, instanceId: p.instanceId })) as any
      })
      const live = produced.filter((p: any) =>
        rows.some(r => r.instanceId === p.instanceId && r.status === 'running' && r.restartPolicy === 'resync')
      )
      if (!live.length) continue

      let sites: SiteDescriptor[]
      try {
        sites = await adapter.discoverSites(ref.connectionConfig ?? {})
      } catch (e: any) {
        /* 원 시스템이 아직 안 떴을 수 있다 — 지어내지 않고 그렇게 적는다(트윈은 계속 「도는 미러」다). */
        for (const p of live) failed.push({ instanceId: p.instanceId, error: `discover failed: ${e?.message ?? 'unknown'}` })
        continue
      }

      for (const p of live) {
        const site = sites.find(s => s.siteId === p.siteId)
        if (!site) {
          failed.push({ instanceId: p.instanceId, error: `site "${p.siteId}" no longer discoverable` })
          continue
        }
        const owner = attached.get(p.instanceId)
        if (owner) {
          /* 두 접속이 같은 트윈을 주장한다 — 먼저 붙은 것을 지키고, 그 사실을 남긴다(둘 다 붙이면 미러가 리셋된다). */
          failed.push({
            instanceId: p.instanceId,
            error: `also claimed by reference "${ref.source}" — kept the feed from "${owner}" (two connections feeding one twin would reset the mirror)`
          })
          continue
        }
        attached.set(p.instanceId, ref.source)
        try {
          const ok = await startReferenceLiveFeed(domainId, ref.adapterType, ref.connectionConfig ?? {}, site, p.instanceId, ref.source)
          if (ok) resumed.push(p.instanceId)
          else failed.push({ instanceId: p.instanceId, error: 'adapter has no openLiveFeed' })
        } catch (e: any) {
          failed.push({ instanceId: p.instanceId, error: e?.message ?? 'reattach failed' })
        }
      }
    }
  } catch (e: any) {
    twinError('[twin-live] feed resume scan failed', e?.message)
  }

  if (resumed.length) twinLog(`[twin-live] reattached ${resumed.length} feed(s): ${resumed.join(', ')}`)
  /* **알리지 않고 넘기지 않는다** — 붙지 않은 미러는 「도는데 아무것도 오지 않는」 트윈이다. */
  for (const f of failed) twinWarn(`[twin-live] "${f.instanceId}" feed NOT reattached — ${f.error}`)
  return { resumed, failed }
}

/*
 * 이 계층의 피드 장부를 등록한다 — 화면이 「끊김」을 말할 근거는 **묻는 자리 하나**여야 한다.
 * (프리셋 기반 피드는 앱이 자기 장부를 따로 등록한다. 한쪽만 보면 흐르는 트윈을 끊겼다고 부른다.)
 */
registerLiveFeedProbe(() => [...feeds.keys()])

/* 인스턴스 정지 시 피드 자동 정리(생명주기 대칭). */
TwinEngine.onStop(instanceId => stopReferenceLiveFeed(instanceId))