import { twinLog } from '../../engine/log.js'
import { hydratedDate } from '../../engine/read-time.js'
import { Arg, Ctx, Int, Mutation, Query, Resolver, Directive, UseMiddleware } from 'type-graphql'

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

import { TwinEngine } from '../../engine/index.js'
import { readSpecCoverage } from '../../engine/spec-coverage.js'
import { describeModelBasis, modelBasisChanged, type ModelBasis } from '../../engine/model-basis.js'
import { TwinEvent } from '../twin-event/twin-event.js'
import { GAP_SAMPLES, aggregatePoints, computeSpc, gapMetrics, learnedCalibration, type GapPoint } from './gap-analytics.js'
import { FORECAST_METRICS, type MetricFn } from './forecast-metrics.js'
import { forecastStepMs } from './forecast-tuning.js'

/* 커널 예측 함수 — CJS require(twin-engine 과 동일 방식). */
const { monteCarloForecastAsync } = require('@operato/twin-kernel')

/**
 * 이벤트 루프에 자리를 내주는 방법 — **커널이 아니라 여기가 안다.**
 *
 * `setImmediate` 는 대기 중인 I/O 다음 차례에 돌아온다. 마이크로태스크(`Promise.resolve()`)로는
 * 자리가 가지 않는다 — 그러면 예측이 도는 동안 라이브 틱도 HTTP 도 여전히 멈춘다.
 *
 * ── 내준 시간은 **부하가 아니다** ───────────────────────────────────────────
 * 부하 계기는 「이 트윈이 루프를 얼마나 붙잡았나」를 재는 자다. 양보를 넣은 뒤로 벽시계 시간에는
 * 남에게 넘긴 시간이 섞이는데, 그것까지 부하로 적으면 계기가 거짓말을 한다 — 하필 이 문제를 짚어낸
 * 그 계기다. 그래서 내준 시간을 따로 세어 빼고 적는다.
 */
function breather() {
  let yieldedMs = 0
  return {
    fn: () =>
      new Promise<void>(resolve => {
        const t = performance.now()
        setImmediate(() => {
          yieldedMs += performance.now() - t
          resolve()
        })
      }),
    get ms() {
      return yieldedMs
    }
  }
}

/*
 * 트윈 예측 채널 — 상향 query. 트윈의 "예측(predict)" 면(설계 §3.5, ADR-0009).
 * 커널 monteCarloForecast(현재에서 N fork → 지표 분포) + 단일 fork 예측 궤적. 원본 live 무간섭(fork 격리).
 * metric 은 도메인-일반(스냅샷 파생), scenario 는 미래 자극(기본=계속 입고, 실 시스템은 인스턴스/설정 파생).
 */
const GTIN = 'urn:epc:idpat:sgtin:0614141.012345.*'
const DEFAULT_SCENARIO = {
  seed: 7,
  generators: [
    { kind: 'inbound-arrival', rate: { distribution: 'poisson', meanPerHour: 60 }, content: { skuMix: [{ gtin: GTIN, weight: 1 }], qtyPerLine: { min: 4, max: 12 } } }
  ]
}
/*
 * 지표 표는 `forecast-metrics.ts` 가 소유한다 — 여기 있던 것을 옮겼다(시험이 이 파일을 불러올 수 없다).
 *
 * **모르는 지표를 `items` 로 갈음하지 않는다.** 예전에는 `METRICS[metric] ?? METRICS.items` 였다.
 * 에너지 트윈에서 `peakKW` 를 물으면 재고 곡선이 그려지고, 재고가 없으니 **0 이 평평하게** 나왔다 —
 * 화면은 그것을 「피크가 0」으로 읽는다. 지어낸 답보다 모른다고 말하는 편이 낫다.
 */
function metricFn(metric?: string): MetricFn | null {
  if (!metric) return FORECAST_METRICS.items
  return FORECAST_METRICS[metric] ?? null
}

/* 학습된 보정 저장(P3.2, 인메모리) — key=domain:instance:metric:horizon → {biasShift, halfWidth, …}.
   twinLearnCalibration 이 갭에서 학습해 저장하고, twinForecast 가 있으면 예측에 상시 적용(라이브 보정).
   데모 트윈은 재부팅 시 저널 리셋이라 세션 스코프로 충분 — DB 영속(TwinForecastScore)은 후속. */
const CALIBRATION = new Map<
  string,
  { biasShift: number; halfWidth: number; at: string; samples: number; accBefore: number | null; accAfter: number | null; basis?: ModelBasis }
>()
const calKey = (d: string, i: string, m: string, h: number) => `${d}:${i}:${m}:${h}`

/**
 * 이 트윈이 백테스트할 수 있는 구간 — [첫 이벤트, 마지막 이벤트 − H].
 * 지평 H 만큼 앞을 봐야 "그때 예측한 것"과 "실제로 벌어진 것"을 맞대 볼 수 있으므로 끝에서 H 를 뺀다.
 */
async function gapWindow(domainId: string, instanceId: string, H: number): Promise<{ t0: number; lastVantage: number } | null> {
  /*
   * ── **양 끝을 집계로 묻는다** (2026-08-28) ──────────────────────────────────
   *
   * 예전에는 `findOne({ where: { domain: { id } }, order: { eventTime } })` 두 번이었다. 관계를 가진
   * 엔티티에서 `findOne` 은 두 단계 질의가 되고 **안쪽에 상한이 없다** — 그 트윈의 저널을 전부 만든 뒤
   * 한 줄을 고른다. 승화푸드는 337만 행이고, 그것을 두 번 했다.
   *
   * `MIN`/`MAX` 는 `(domain, instance, event_time)` 색인의 양 끝을 집으므로 각각 한 줄이다.
   * 이 파일의 다른 자리와 같은 규율이다(§`twin-engine` 의 `tipOf`).
   */
  const repo = getRepository(TwinEvent)
  const edge = (agg: 'MIN' | 'MAX') =>
    repo
      .createQueryBuilder('e')
      .select(`${agg}(e.eventTime)`, 'at')
      .where('e.domain = :domainId', { domainId })
      .andWhere('e.instanceId = :instanceId', { instanceId })
      .getRawOne<{ at: unknown }>()
  const [firstRow, lastRow] = await Promise.all([edge('MIN'), edge('MAX')])
  /* 드라이버가 준 값은 시간대가 없을 수 있다 — 컬럼 변환을 지나게 한다(§`hydratedDate`). */
  const first = hydratedDate(repo, 'eventTime', firstRow?.at)
  const last = hydratedDate(repo, 'eventTime', lastRow?.at)
  if (!first || !last) return null
  const t0 = first.getTime()
  const lastVantage = last.getTime() - H
  return lastVantage > t0 ? { t0, lastVantage } : null
}

/**
 * 갭 point 계산(공용) — 주어진 vantage 각각에서 forecast-at-T vs actual-at-T+H(결정적 백테스트).
 *
 * **vantage 는 밖에서 준다.** 예전에는 이 함수가 자기 저널의 첫·마지막 이벤트로 격자를 스스로 만들었다.
 * 그러면 트윈마다 격자가 달라져, 공동배치 합산에서 `points[3]` 끼리 더할 때 **서로 다른 시각의 값**이
 * 더해졌다(자세한 사연은 `aggregatePoints` 주석).
 *
 * 반환 배열은 격자와 **같은 길이**이고, 그 vantage 를 재지 못했으면 그 자리에 `null` 을 둔다.
 * 건너뛰고 밀어 담으면 배열 인덱스가 시간축이기를 그만둔다.
 */
async function gapPoints(domainId: string, instanceId: string, m: (s: any) => number, H: number, grid: number[], scen: any): Promise<(any | null)[]> {
  const points: (any | null)[] = []
  for (const T of grid) {
    const k = await TwinEngine.buildForecastKernelAt(domainId, instanceId, new Date(T).toISOString()).catch(() => null)
    if (!k) { points.push(null); continue }
    const dist = await monteCarloForecastAsync(k, { runs: 12, horizonMs: H, scenario: scen, tickMs: forecastStepMs(H), metric: m, yieldFn: breather().fn })
    const naive = m(k.getSnapshot())
    const st = await TwinEngine.recover(domainId, instanceId, undefined, new Date(T + H).toISOString()).catch(() => null)
    if (!st) { points.push(null); continue }
    const a = m(st)
    const p50 = dist?.p50 ?? naive
    points.push({ t: T, p50, min: dist?.min ?? p50, max: dist?.max ?? p50, actual: a, error: a - p50, inBand: a >= (dist?.min ?? -Infinity) && a <= (dist?.max ?? Infinity), naiveAbsError: Math.abs(a - naive), modelAbsError: Math.abs(a - p50) })
  }
  return points
}

/**
 * 여러 트윈이 **함께** 백테스트할 수 있는 구간과 그 위의 vantage 격자.
 *
 * 겹치는 구간은 각자의 구간을 교집합한 것 — 시작은 가장 늦은 t0, 끝은 가장 이른 lastVantage.
 * 한 트윈이라도 그 시각에 저널이 없으면 공장 총량을 말할 수 없기 때문이다.
 * 겹치는 구간이 없으면 `null` — "0" 이 아니라 "잴 수 없음" 이다.
 */
function sharedGrid(windows: { t0: number; lastVantage: number }[], N: number): number[] | null {
  if (!windows.length || N < 2) return null
  const t0 = Math.max(...windows.map(w => w.t0))
  const lastVantage = Math.min(...windows.map(w => w.lastVantage))
  if (!(lastVantage > t0)) return null
  return Array.from({ length: N }, (_, i) => t0 + ((lastVantage - t0) * i) / (N - 1))
}

@Resolver()
@UseMiddleware(SpeakRefusals)
export class TwinForecastQuery {
  @Query(returns => ScalarObject, {
    nullable: true,
    description: 'Monte-Carlo forecast of a metric over a horizon (twin prediction faculty) + a single-fork predicted trajectory. Live instance is untouched (forks are isolated).'
  })
  async twinForecast(
    @Arg('instanceId') instanceId: string,
    @Arg('metric', { nullable: true }) metric: string,
    @Arg('runs', type => Int, { nullable: true }) runs: number,
    @Arg('horizonMs', type => Int, { nullable: true }) horizonMs: number,
    @Arg('scenario', type => ScalarObject, { nullable: true }) scenario: any,
    @Arg('at', { nullable: true }) at: string,
    @Ctx() context: ResolverContext
  ): Promise<any> {
    /* vantage(at) 기준 예측 — 히스토리 노브가 과거 T 면 "그때 서서 본 미래"(백테스트).
       at 있으면 T 시점 커널을 저널로 재구성한다(그건 우회가 아니라 시간여행의 본질).
       없으면 **지금 그 커널**을 쓴다 — sim 이든 live 든 인스턴스가 커널이다(통합 P2). */
    const domainId = context.state.domain.id
    const k = at
      ? await TwinEngine.buildForecastKernelAt(domainId, instanceId, at)
      : TwinEngine.owns(domainId, instanceId)
        ? TwinEngine.kernel(domainId, instanceId)
        : undefined
    if (!k) return null

    const m = metricFn(metric)
    /* 이 트윈이 못 세는 지표 — 빈 예측이 아니라 「그 지표는 여기 없다」고 말한다. */
    if (!m) return { metric, unsupportedMetric: true, distribution: null, trajectory: [] }
    const H = horizonMs ?? 300000
    const R = runs ?? 30
    /*
     * 틱 크기는 **지평선이 정한다**(`forecast-tuning`). 1초 고정이던 시절 5분 지평선이 회차마다
     * 300틱이었고, 30회차면 9,000틱·28.5초였다 — 실측으로 5초 단위와 답이 같다.
     */
    const step = forecastStepMs(H)
    const scen = scenario ?? DEFAULT_SCENARIO

    /*
     * 분포: N개 확률적 미래 표본(fork 마다 seed 변주, 원본 보존).
     *
     * **이것이 요청 처리 중 메인 이벤트 루프에서 동기로 수행된다.** 기본값으로 fork 30개 × 300틱 = 9,000틱.
     * 계기판에 남겨 두지 않으면 "왜 서버가 굳었나" 를 나중에 짚을 수 없다(실제로 한 번 겪었다).
     */
    const tMc = performance.now()
    const mcBreath = breather()
    const distribution = await monteCarloForecastAsync(k, { runs: R, horizonMs: H, scenario: scen, tickMs: step, metric: m, yieldFn: mcBreath.fn })
    TwinEngine.recordLoad(domainId, instanceId, 'forecast', performance.now() - tMc - mcBreath.ms)
    TwinEngine.countForks(domainId, instanceId, R)

    /* 학습된 보정 상시 적용(P3.2) — twinLearnCalibration 이 저장한 편향·밴드가 있으면 예측을 현실로 보정(라이브가 실제로 나아짐).
       갭트렌드(gapPoints)는 raw 모델을 그대로 측정(학습 신호 보존) — 보정은 표시용 forecast 에만. */
    const stored = CALIBRATION.get(calKey(domainId, instanceId, metric ?? 'items', H))
    /*
     * **배운 모델에만 적용한다.** 보정은 "그때 모델이 이만큼 치우쳐 있다" 를 배운 것이다. 소요시간
     * 추정기(실측·거리)나 공정 명세가 바뀌어 모델이 달라지면 그 편향은 더 이상 사실이 아니고, 그대로
     * 밀어 주면 이중 보정이 된다. 그래서 지문을 비교하고, 다르면 적용하지 않되 **그 사실을 응답에
     * 싣는다** — 오류 없이 사라지면 사용자는 학습이 왜 안 듣는지 알 수 없다(다시 배우면 된다).
     * 지문을 알 수 없을 때는 비교하지 않는다(모르는 것을 근거로 보정을 끊지 않는다).
     */
    const basis = await TwinEngine.modelBasis(domainId, instanceId)
    const stale = !!stored && modelBasisChanged(stored.basis, basis)
    if (stale) {
      twinLog(
        `[twin-forecast] "${instanceId}": calibration not applied — model changed since it was learned.` +
          ` learned on: ${describeModelBasis(stored!.basis)} / now: ${describeModelBasis(basis)}`
      )
    }
    /*
     * 분위수 — 표본에서 직접 뽑는다.
     *
     * 커널이 표본 배열을 그대로 돌려주므로(forecast.ts) 여기서 정렬해 p05·p95 를 낸다. 이것이 없으면
     * 화면이 보여 줄 수 있는 폭은 관측 **최소·최대** 뿐인데, 그건 신뢰구간이 아니라 극단값 두 개다.
     * "운이 나쁜 5% 에서는 이만큼" 이라고 말하려면 분위수여야 한다. 표본이 적으면(runs 가 작으면)
     * 분위수도 거칠다 — 그래서 표본 수를 함께 싣는다(sampleCount). 숫자만 주고 근거를 감추지 않는다.
     */
    if (distribution) {
      const samples: number[] = Array.isArray(distribution.samples) ? distribution.samples : []
      if (samples.length) {
        const sorted = [...samples].sort((a, b) => a - b)
        const pct = (p: number) => sorted[Math.min(sorted.length - 1, Math.floor(p * sorted.length))]
        distribution.p05 = pct(0.05)
        distribution.p95 = pct(0.95)
      }
      distribution.sampleCount = samples.length
    }

    const cal = stored && !stale ? stored : undefined
    if (cal && distribution) {
      /*
       * `p50` 은 커널 계약상 항상 있다(`MonteCarloResult.p50: number`). 그래도 메우지 않는다 —
       * 예전에는 `?? 0` 이었는데, 만약 비면 보정된 p50 이 **편향값 그 자체**가 되고 min·max 는
       * 그 둘레로 그려진다. 예측이 아닌 숫자가 예측의 자리에 앉는데 아무 신호가 없다.
       */
      if (typeof distribution.p50 !== 'number') throw new Error('forecast distribution has no p50 — kernel contract broken')
      const p50c = distribution.p50 + cal.biasShift
      distribution.p50 = Math.round(p50c)
      distribution.min = Math.round(p50c - cal.halfWidth)
      distribution.max = Math.round(p50c + cal.halfWidth)
      /*
       * 분위수는 **옮기기만** 한다. 보정이 배운 것은 "모델이 이만큼 치우쳐 있다"(위치)이지 분포의
       * 모양이 아니다. p50 처럼 밀어 두어야 p05~p95 폭이 표본이 말한 그대로 남는다.
       * (min·max 는 위에서 학습 밴드로 대체된다 — 보정이 걸리면 그 둘의 뜻이 '관측 극단' 에서
       *  '학습 밴드' 로 바뀐다는 뜻이다. 화면이 둘을 같은 것으로 그리면 안 된다.)
       */
      if (distribution.p05 !== undefined) distribution.p05 = Math.round(distribution.p05 + cal.biasShift)
      if (distribution.p95 !== undefined) distribution.p95 = Math.round(distribution.p95 + cal.biasShift)
      if (distribution.p90 !== undefined) distribution.p90 = Math.round(distribution.p90 + cal.biasShift)
      distribution.calibrated = true
    }

    /*
     * 예측 궤적(예측선): 단일 fork 를 horizon 까지 실행하며 지표 기록. 이것도 잰다.
     *
     * 여기도 자리를 내준다 — 실측으로 이 한 회차가 1.3초였다(300틱 + 스냅샷 300번). 분포만큼은
     * 아니지만, 라이브 틱 간격(1초)보다 길게 붙잡으면 그 사이 트윈들이 한 박자씩 밀린다.
     */
    const tTraj = performance.now()
    const trajBreath = breather()
    const fc = k.fork()
    fc.scenario.load({ ...scen, seed: 1 })
    fc.scenario.start()
    const trajectory: { t: number; value: number }[] = []
    for (let t = 0; t < H; t += step) {
      fc.tick(step)
      trajectory.push({ t: t + step, value: m(fc.getSnapshot()) })
      /* 틱마다 내주면 양보 자체가 비용이 된다 — 라이브 틱 간격보다 훨씬 짧은 단위로 끊는다. */
      if (trajectory.length % 25 === 0) await trajBreath.fn()
    }

    TwinEngine.recordLoad(domainId, instanceId, 'forkTick', performance.now() - tTraj - trajBreath.ms)
    TwinEngine.countForks(domainId, instanceId, 1)

    const current = m(k.getSnapshot()) // vantage T 의 실제값(= naive '그대로 유지' 예측의 기준)

    /* 실측 오버레이 + 채점(P1) — vantage 가 과거(at)일 때만: T 이후 실제로 일어난 metric 을 T→min(now,T+H) 로 샘플.
       현재 이후는 실측이 없음(미래). recover(T+t) 로 각 샘플 상태 재구성해 metric 추출. 비용=샘플 수 만큼 replay. */
    const actual: { t: number; value: number }[] = []
    let scored: any = null
    if (at) {
      const T = Date.parse(at)
      const endMs = Math.min(Date.now(), T + H)
      const span = endMs - T
      if (Number.isFinite(T) && span > 0) {
        const SAMPLES = 8
        const dt = span / SAMPLES
        for (let i = 1; i <= SAMPLES; i++) {
          const st = await TwinEngine.recover(domainId, instanceId, undefined, new Date(T + i * dt).toISOString()).catch(() => null)
          if (st) actual.push({ t: i * dt, value: m(st) })
        }
        /* 지평선이 지났으면(T+H ≤ now) 그 지점 실측 vs 예측 채점(단일 vantage). p50 대비 오차·밴드 적중·FVA(naive=현재값 유지 대비). */
        if (T + H <= Date.now() && actual.length) {
          const a = actual[actual.length - 1].value
          const p50 = distribution?.p50 ?? current
          scored = {
            actual: a,
            p50,
            error: a - p50, // 부호 = 편향 방향(+ 실제가 더 많음)
            inBand: a >= (distribution?.min ?? -Infinity) && a <= (distribution?.max ?? Infinity),
            modelAbsError: Math.abs(a - p50),
            naiveAbsError: Math.abs(a - current) // naive = "그대로 유지"(현재값). FVA = naive - model
          }
        }
      }
    }

    /*
     * 시뮬 명세 자기보고를 함께 낸다 — **상수로 계산한 예측과 현장 명세로 계산한 예측이 화면에서 같아
     * 보이지 않게.** 코드·수치만 싣고 사람이 읽는 문장은 클라가 만든다(언어 중립).
     * 커널이 이 API 를 갖지 않은 버전이면 키를 만들지 않는다(모르는 것을 "전부 기본값" 이라 단정 금지).
     */
    /* 보고는 **실제로 실행한 fork** 에서 읽는다 — 부모 커널(k)은 tick 하지 않으므로(예측은 fork 에서
     * 수행된다) 부모에서 읽으면 씨앗으로 세운 예측에서는 늘 비어 있다. 숫자를 낸 주체가 보고해야 한다. */
    const specCoverage = readSpecCoverage(fc) ?? readSpecCoverage(k)
    return {
      metric: metric ?? 'items', horizonMs: H, runs: R, current, distribution, trajectory, actual, scored,
      /* 얼마나 잘게 굴렸나 — 숫자만 주고 근거를 감추지 않는다(같은 규율로 `specCoverage` 도 싣는다). */
      tickMs: step,
      /*
       * **가정이 어떻게 됐나** — 걸린 개입·거절된 개입·덮어쓴 선언.
       *
       * 이것을 감추면 두 가지가 들어온 것이 없어진다: 걸리지 않은 가정(「걸었는데 왜 안 바뀌나」)과, 덮어쓴
       * 선언(사용자가 what-if 결과를 트윈의 현재 상태로 읽는다). 실제로 실행한 fork(`fc`)에서 읽는다 —
       * 가정을 겪은 것은 그쪽이다.
       */
      ...(typeof (fc as any).interventionOutcomes === 'function'
        ? (() => {
            const outcomes = (fc as any).interventionOutcomes()
            return outcomes.length ? { interventions: outcomes } : {}
          })()
        : {}),
      ...(typeof (fc as any).declarationOverrides === 'function'
        ? (() => {
            const overrides = (fc as any).declarationOverrides()
            return overrides.length ? { overrides } : {}
          })()
        : {}),
      calibrated: !!cal,
      /* 학습해 뒀지만 모델이 바뀌어 적용하지 않았다 — 사용자가 다시 배울 수 있게 알린다. */
      ...(stale ? { calibrationStale: true } : {}),
      ...(specCoverage ? { specCoverage } : {}),
      /*
       * **계보에 구멍이 몇 칸 생겼나** — 개체를 잇지 못한 채 넘어간 공정 수.
       *
       * ── 왜 이 값을 예측 응답에 싣나 (2026-08-24) ──────────────────────────────
       * 미러의 오더는 확보분을 갖지 않는다(원본이 「어느 개체가 잡혀 있나」를 말하지 않는 것이 그
       * 시스템의 정상이다). 커널은 그 자리에서 **던지지 않고 넘긴다** — 단계를 넘는 것은 공정의
       * 진행이고 개체 번호가 필요 없다. 대신 넘긴 횟수를 센다.
       *
       * 그런데 그 카운터는 **fork 안에** 있고 fork 는 버려진다. 여기서 싣지 않으면 아무에게도 닿지
       * 않는다 — 리듀서가 오래 세어 온 값을 스냅샷이 떨어뜨려 그것을 읽도록 쓰인 검사가 영원히
       * 아무 신호도 내지 않았던 것과 **같은 실수**다. 세어 놓고 내보내지 않으면 안 센 것과 같다.
       *
       * 왜 화면이 이것을 알아야 하나: 계보는 **회수 범위**를 정하는 사슬이다. 구멍이 있는데 온전한
       * 것처럼 보이면 회수가 알리지 않고 좁아진다. 식품에서 그것은 되돌릴 수 없다.
       *
       * 두 곳에서 읽어 **큰 값**을 낸다 — 분포(회차 R개의 최대)와 궤적 fork 는 **둘 다 실제로 실행됐다**.
       * 한쪽만 보면 다른 쪽이 겪은 구멍을 잃는다.
       */
      ...(() => {
        const gaps = Math.max(
          (distribution as any)?.stepsWithoutMaterial ?? 0,
          (fc as any)?.getSnapshot?.()?.stepsWithoutMaterial ?? 0
        )
        /* 0 은 싣지 않는다 — 「구멍이 없다」와 「이 축을 모른다」를 화면이 구별할 수 있게. */
        return gaps > 0 ? { stepsWithoutMaterial: gaps } : {}
      })()
    }
  }

  /*
   * 갭 트렌드(P2) — N개 과거 vantage 각각에서 "그때 예측 vs 그 지평선의 실제"를 재계산(결정적 백테스트).
   * 지평선이 이미 지난 vantage 만([t0, tLast−H]) → actual 확정. 온디맨드 재계산(누적/영속은 긴 기간용 후속 최적화).
   * 비용 = N × (buildForecastKernelAt replay + monteCarlo + recover replay). N·runs 를 낮게 bound. per-instance(클라가 co-located 합산).
   */
  @Query(returns => ScalarObject, {
    nullable: true,
    description:
      'Forecast-accuracy gap diagnosis (canonical, server-computed): for N past vantages recompute forecast-at-T vs actual-at-T+h (deterministic backtest), aggregate co-located instances, and return points + SPC verdict (stability/bias) + metrics (accuracy/coverage/FVA) + learned-calibration projection. Client only renders.'
  })
  async twinForecastGapTrend(
    @Arg('instanceId') instanceId: string,
    @Arg('metric', { nullable: true }) metric: string,
    @Arg('horizonMs', type => Int, { nullable: true }) horizonMs: number,
    @Arg('samples', type => Int, { nullable: true }) samples: number,
    @Arg('scenario', type => ScalarObject, { nullable: true }) scenario: any,
    @Ctx() context: ResolverContext,
    @Arg('instanceIds', type => [String], { nullable: true }) instanceIds?: string[]
  ): Promise<any> {
    const domainId = context.state.domain.id
    const m = metricFn(metric)
    const H = horizonMs ?? 300000
    if (!m) return { metric, unsupportedMetric: true, horizonMs: H, samples: 0, points: [], spc: computeSpc([]), metrics: null, learn: null }
    const N = Math.min(30, Math.max(2, samples ?? GAP_SAMPLES)) // 표본은 서버가 관장(SPC 신뢰 하한 ~20).
    const scen = scenario ?? DEFAULT_SCENARIO
    // 공동배치(co-located) 집계도 서버에서 — 공장 단위 진단이 canonical. 미지정 시 단건.
    const ids = instanceIds?.length ? instanceIds : [instanceId]

    /*
     * 격자를 **먼저 하나** 만든다. 트윈마다 자기 저널로 격자를 만들면 같은 인덱스가 서로 다른
     * 시각이 되어, 합산이 다른 시점의 값을 더한다(`aggregatePoints` 주석 참조).
     *
     * 저널이 없거나 지평 H 를 담을 만큼 길지 않은 트윈은 **백테스트에 참여할 수 없다.** 그런 트윈은
     * 빼되 누구를 뺐는지 돌려준다 — 알리지 않고 빼면 화면은 그 트윈까지 포함한 공장 총량으로 읽는다.
     */
    const windows = await Promise.all(ids.map(async id => ({ id, w: await gapWindow(domainId, id, H) })))
    const usable = windows.filter(x => x.w)
    const excluded = windows.filter(x => !x.w).map(x => x.id)
    const grid = sharedGrid(usable.map(x => x.w!), N)
    if (!grid) {
      /* 겹치는 구간이 없다 — 잴 수 없는 것이지 정확도가 0 인 것이 아니다. */
      return { metric: metric ?? 'items', horizonMs: H, samples: N, points: [], spc: computeSpc([]), metrics: null, learn: null,
        coverage: { instances: ids.length, measured: 0, excluded, vantages: N, used: 0, dropped: 0, reason: 'no-common-window' } }
    }

    const per = await Promise.all(usable.map(x => gapPoints(domainId, x.id, m, H, grid, scen)))
    const { points, droppedVantages } = aggregatePoints(per as (GapPoint | null)[][])
    const metrics = points.length ? gapMetrics(points) : null
    const spc = computeSpc(points.map(p => p.error))
    const learn = metrics ? learnedCalibration(points, metrics) : null
    return { metric: metric ?? 'items', horizonMs: H, samples: N, points, spc, metrics, learn,
      /* 이 숫자가 **무엇으로 계산됐는지**. 24점 중 18점이면 화면이 그렇게 말해야 한다. */
      coverage: { instances: ids.length, measured: usable.length, excluded, vantages: N, used: points.length, dropped: droppedVantages } }
  }

  /*
   * 학습(P3.2) — 갭에서 편향·밴드를 학습해 저장하면, 이후 twinForecast 가 상시 적용(라이브 예측이 실제로 보정됨).
   * assimilation-lite: 잔차(innovation)를 되먹여 추정치를 현실로. per-instance(클라가 co-located 합산).
   * 반환: 학습된 보정 + before/after 정확도(효과 정량화). 저장은 인메모리(세션) — DB 영속은 후속.
   */
  @Directive('@privilege(category: "twin", privilege: "mutation", domainOwnerGranted: true, superUserGranted: true)')
  @Mutation(returns => ScalarObject, { description: 'Learn a forecast calibration (bias + band) from the gap trend and persist it so future forecasts apply it. Returns the calibration + before/after accuracy.' })
  async twinLearnCalibration(
    @Arg('instanceId') instanceId: string,
    @Arg('metric', { nullable: true }) metric: string,
    @Arg('horizonMs', type => Int, { nullable: true }) horizonMs: number,
    @Ctx() context: ResolverContext,
    @Arg('clear', type => Boolean, { nullable: true }) clear?: boolean
  ): Promise<any> {
    const domainId = context.state.domain.id
    if (!TwinEngine.owns(domainId, instanceId)) return { ok: false } // 테넌트 격리
    const m = metricFn(metric)
    if (!m) return { ok: false, reason: 'unsupported-metric', metric }
    const H = horizonMs ?? 300000
    // 보정 해제 — 저장된 calibration 제거(이후 예측은 raw 로 복귀). 학습을 되돌리는 정직한 off 스위치.
    if (clear) {
      CALIBRATION.delete(calKey(domainId, instanceId, metric ?? 'items', H))
      return { ok: true, cleared: true, metric: metric ?? 'items', horizonMs: H }
    }
    // 표본·계산을 gap-analytics 로 일원화(gapTrend 진단과 동일 N·동일 수식 → 카드 예상치와 실제 보정 일치).
    /* 학습은 단건이라 격자도 이 트윈의 구간 하나다. 저널이 지평 H 를 담을 만큼 길지 않으면 배울 것이 없다. */
    const w = await gapWindow(domainId, instanceId, H)
    const grid = w ? sharedGrid([w], GAP_SAMPLES) : null
    if (!grid) return { ok: false, reason: 'insufficient-history' }
    /* 못 잰 vantage 는 `null` 로 온다 — 0 으로 세면 편향 학습이 그만큼 끌려간다. */
    const pts = (await gapPoints(domainId, instanceId, m, H, grid, DEFAULT_SCENARIO)).filter(Boolean) as GapPoint[]
    if (pts.length < 3) return { ok: false, reason: 'insufficient-history' }
    const learn = learnedCalibration(pts, gapMetrics(pts))
    /* 학습한 모델의 지문을 함께 남긴다 — 이후 모델이 바뀌면 이 보정은 적용되지 않는다. */
    const basis = await TwinEngine.modelBasis(domainId, instanceId)
    const cal = {
      biasShift: learn.biasShift,
      halfWidth: learn.halfWidth,
      at: '',
      samples: pts.length,
      accBefore: learn.accBefore,
      accAfter: learn.accAfter,
      ...(basis ? { basis } : {})
    }
    CALIBRATION.set(calKey(domainId, instanceId, metric ?? 'items', H), cal)
    return { ok: true, metric: metric ?? 'items', horizonMs: H, biasShift: cal.biasShift, halfWidth: cal.halfWidth, samples: pts.length, accBefore: cal.accBefore, accAfter: cal.accAfter }
  }
}