track-f Engineer Contract — Decoder Capacity Swap 구현¶

User 결정 pending (5건) — contract dispatch 는 아래 결정 확정 후에만 가능: 1. W1 decoder 구조 (α=FC-MLP / β=Transformer / γ=병렬) — 본문 γ 전제로 W1+W2 둘 다 구현 2. Aggregation 복원 범위 (decoder 단독 / v6 포함) — 본문 decoder-only 전제, FL aggregation 코드 변경 없음 3. rounds 확장 (smoke 수렴 후 조건부) — 본문 runtime 선택사항, 코드 불변 4. Gate 수치 (파일럿 후 확정) — 본문 변수 반영 불필요 5. Gray-zone 정책 — 본문 변수 반영 불필요

본문은 옵션 γ + decoder-only + critic F3 3 보강 (RevIN biased, loss_decode, DLinear differential LR) 전제. user 가 α 또는 β 선택 시 W1 또는 W2 중 하나만 구현.

본 문서는 track-f phase 의 코드 구현을 engineer 에이전트에게 위임하기 위한 계약서이다. 범위, 금지 사항, 테스트 요건, 수용 기준을 명시한다.

0. 단 하나의 목표¶

experiments/federated/v7_runner.py 의 ProposedModel 클래스에 decoder_type 인자를 추가하여 3가지 분기 (v7_conv default, v6_fc W1, tf2_128 W2) 를 지원하고, CellSpec / CELL_REGISTRY 에 W1/W2 를 등록한다. 기존 12 cell (B0~V5) 은 bit-exact backward-compat.

1. 허용 수정 범위¶

수정 가능: experiments/federated/v7_runner.py
새 파일 생성 허용: tests/test_track_f_*.py (필요 시 테스트 파일 분리)
import 허용 (기존 경로):
from lib.models.Encoder_and_Retrieval import Encoder (v6 원본)
from lib.models.decoder import XcodeYtimeDecoder (v6 원본)
단, 이 import 경로가 v7_runner.py 에서 해석되도록 sys.path 조정 또는 init 추가 필요 (기존 v6 experiment 파일이 하는 방식 동일하게)

2. 금지 수정 범위 (strict)¶

src/fed_learning/fedpm.py 수정 금지 (v6 베이스)
src/FedUnit-64D1/lib/** 수정 금지 (v6 원본 라이브러리, import only)
src/peak_analysis/v7/metrics.py 수정 금지 (golden tensor 소스)
기존 CELL_REGISTRY 12종 (B0~V5) 의 동작 변경 금지
v7-planning 브랜치 유지 (새 브랜치 생성 금지)
Git user config 변경 금지
기존 test 삭제/수정 금지 (추가만 허용)

3. 구현 단계¶

3.1 CellSpec 확장¶

v7_runner.py 의 CellSpec dataclass 에 두 필드 추가 (critic F3 반영):

@dataclass
class CellSpec:
    name: str
    method: str
    peak_loss: bool
    use_vq: bool
    use_dlinear: bool
    vq_alignment: bool = False
    vq_reset: bool = False
    decoder_type: str = "v7_conv"       # NEW: default backward-compat
    dlinear_lr_scale: float = 1.0       # NEW: v6 R1b DLinear differential LR (W1/W2 = 0.1)
    extra_hparams: dict = field(default_factory=dict)

dlinear_lr_scale=1.0 default 로 기존 B0~V5 cell 동작 불변 (backward-compat).
W1, W2 cell 에서 dlinear_lr_scale=0.1 설정 → optimizer 생성 시 DLinear 파라미터 group 을 base_lr × 0.1 로 학습 (v6 R1b line 308-313 pattern).

3.2 ProposedModel 확장¶

v7_runner.py:663 의 ProposedModel.__init__ 에 decoder_type kwarg 추가:

def __init__(
    self,
    seq_len: int = SEQ_LEN,
    pred_len: int = PRED_LEN,
    num_embeddings: int = VQ_NUM_EMBEDDINGS,
    embedding_dim: int = VQ_EMBEDDING_DIM,
    commitment_beta: float = VQ_COMMITMENT_BETA,
    use_dlinear: bool = True,
    use_ema: bool = False,
    ema_gamma: float = VQ_EMA_GAMMA_DEFAULT,
    vq_reset: bool = False,
    vq_reset_threshold: int = 2,
    decoder_type: str = "v7_conv",   # NEW
) -> None:
    super().__init__()
    ...
    self.decoder_type = decoder_type

    if decoder_type == "v7_conv":
        # 기존 로직 — 변경 없음
        self.patch_proj = nn.Linear(self.PATCH_LEN, embedding_dim)
        self.encoder = nn.Sequential(
            nn.Conv1d(embedding_dim, embedding_dim, kernel_size=3, padding=1), nn.ReLU(),
            nn.Conv1d(embedding_dim, embedding_dim, kernel_size=3, padding=1), nn.ReLU(),
        )
        # VQ layer 동일 구성
        ...
        self.decoder = nn.Sequential(
            nn.Conv1d(embedding_dim, embedding_dim, kernel_size=3, padding=1), nn.ReLU(),
            nn.Conv1d(embedding_dim, embedding_dim, kernel_size=3, padding=1), nn.ReLU(),
        )
        self.pred_head = nn.Linear(embedding_dim, 1)

    elif decoder_type == "v6_fc":
        # W1: v6 R1b 1:1 복원
        from lib.models.Encoder_and_Retrieval import Encoder as V6Encoder
        from lib.models.decoder import XcodeYtimeDecoder

        self.patch_proj = None  # v6 Encoder는 채널=1 직접 입력

        self.encoder = V6Encoder(
            in_channels=1,
            num_hiddens=128,
            num_residual_layers=2,
            num_residual_hiddens=64,
            embedding_dim=embedding_dim,
            compression_factor=4,
        )

        # VQ layer 동일 (v7 VectorQuantizer 유지)
        ...

        self.decoder = XcodeYtimeDecoder(
            d_in=embedding_dim,
            d_model=64,
            nhead=4,
            d_hid=256,
            nlayers=4,
            seq_in_len=seq_len // 4,     # 96 // 4 = 24
            seq_out_len=pred_len,        # 24
            dropout=0.0,
            decoder_type='fc',
            compression_factor=4,
            num_residual_layers=2,
            num_residual_hiddens=64,
        )
        self.pred_head = None  # XcodeYtimeDecoder이 Linear(seq_in_len*d_model, pred_len) 내장

    elif decoder_type == "tf2_128":
        # W2 fallback: Transformer 2-layer
        self.patch_proj = nn.Linear(self.PATCH_LEN, embedding_dim)
        self.encoder = nn.Sequential(
            nn.Conv1d(embedding_dim, embedding_dim, kernel_size=3, padding=1), nn.ReLU(),
            nn.Conv1d(embedding_dim, embedding_dim, kernel_size=3, padding=1), nn.ReLU(),
        )
        # VQ layer 동일
        ...
        self.decoder = TransformerDecoderW2(
            d_model=embedding_dim, nhead=4, d_hid=128, nlayers=2,
            seq_len=self.num_patches, pred_len=pred_len,
        )
        self.pred_head = None   # TransformerDecoderW2 이 out_proj 내장

    else:
        raise ValueError(f"Unknown decoder_type: {decoder_type}")

    if use_dlinear:
        self.dlinear = DLinear(seq_len=seq_len, pred_len=pred_len, channels=1)
    else:
        self.dlinear = None

3.3 Forward pass 분기¶

현 forward() 는 v7_conv 전용. 분기 로직 추가:

def forward(self, x: torch.Tensor) -> tuple[torch.Tensor, torch.Tensor, float]:
    if self.decoder_type == "v7_conv":
        return self._forward_v7_conv(x)
    elif self.decoder_type == "v6_fc":
        return self._forward_v6_fc(x)
    elif self.decoder_type == "tf2_128":
        return self._forward_tf2_128(x)

def _forward_v7_conv(self, x):
    # 기존 forward() 내용 그대로 이동
    ...

def _forward_v6_fc(self, x):
    # x: [B, 96, 1]
    # CRITIC F3 보강 1: RevIN affine=False, unbiased=False (biased std) — v6 R1b 원본 일치
    # PyTorch default .std() 는 unbiased=True (N-1) 이므로 v6 와 불일치. var(unbiased=False) 로 재현.
    x_sq = x.squeeze(-1)                                              # [B, 96]
    mean = x_sq.mean(dim=1, keepdim=True)
    var = x_sq.var(dim=1, keepdim=True, unbiased=False)               # biased std
    std = var.sqrt() + 1e-8
    x_norm = (x_sq - mean) / std

    # v6 Encoder 는 [B, 1, 96] 입력
    x_nchw = x_norm.unsqueeze(1)    # [B, 1, 96]
    z = self.encoder(x_nchw)        # [B, D=64, seq=24] (compression_factor=4)

    # VQ: expects [B, L, D]
    z = z.transpose(1, 2)           # [B, 24, 64]
    z_hat, vq_loss, codebook_util = self.vq(z)

    # XcodeYtimeDecoder: expects [seq, B, D] (batch_first=False)
    # CRITIC F3 보강 2: decoder 가 y_hat + y_decode_intermediate tuple return
    # → loss_decode auxiliary supervision 경로 지원
    z_hat_tdb = z_hat.transpose(0, 1)                                 # [24, B, 64]
    y_vq_norm, y_decode_intermediate = self.decoder(z_hat_tdb)        # tuple
    # y_vq_norm: [B, 24]   normalized final pred
    # y_decode_intermediate: [B, 24]   decoder intermediate (normalized) for loss_decode

    # De-normalize (RevIN inverse)
    y_vq = y_vq_norm * std + mean        # [B, 24]

    # DLinear residual (local, non-shared)
    # 주의: DLinear 는 normalized input 아닌 raw x 를 받음 (v6 R1b 동일 동작)
    if self.dlinear is not None:
        y_dlinear = self.dlinear(x).squeeze(-1)   # [B, 24]
        y_hat = y_vq + y_dlinear
    else:
        y_hat = y_vq

    # aux dict 반환: training loop 에서 loss_decode 계산 시 사용
    aux = {
        "y_decode_intermediate": y_decode_intermediate,
        "x_norm": x_norm,                 # loss_decode target = x_norm 또는 target_norm
    }
    return y_hat.unsqueeze(-1), vq_loss, codebook_util, aux

def _forward_tf2_128(self, x):
    # CRITIC F3 보강 1: RevIN biased std (v6 convention 과 동일 처리)
    x_sq = x.squeeze(-1)
    mean = x_sq.mean(dim=1, keepdim=True)
    var = x_sq.var(dim=1, keepdim=True, unbiased=False)
    std = var.sqrt() + 1e-8
    x_norm = (x_sq - mean) / std

    B = x_norm.shape[0]
    patches = x_norm.reshape(B, self.num_patches, self.PATCH_LEN)
    z = self.patch_proj(patches)     # [B, 24, D]
    z = z.transpose(1, 2)            # [B, D, 24]
    z = self.encoder(z)              # [B, D, 24]
    z = z.transpose(1, 2)            # [B, 24, D]

    z_hat, vq_loss, codebook_util = self.vq(z)

    # CRITIC F3 보강 2: TransformerDecoderW2 도 tuple return 으로 확장
    # → (y_vq_norm, y_decode_intermediate). intermediate = pre-out_proj representation
    y_vq_norm, y_decode_intermediate = self.decoder(z_hat)   # [B, 24], [B, 24]

    y_vq = y_vq_norm * std + mean

    if self.dlinear is not None:
        y_dlinear = self.dlinear(x).squeeze(-1)
        y_hat = y_vq + y_dlinear
    else:
        y_hat = y_vq

    aux = {
        "y_decode_intermediate": y_decode_intermediate,
        "x_norm": x_norm,
    }
    return y_hat.unsqueeze(-1), vq_loss, codebook_util, aux

3.3a Forward return signature 변경 (critic F3 보강 2)¶

W1/W2 는 (y, vq_loss, util, aux) 4-tuple 반환. 기존 v7_conv 경로는 backward-compat 유지를 위해 3-tuple + optional aux 로 확장:

def forward(self, x):
    if self.decoder_type == "v7_conv":
        y, vq_loss, util = self._forward_v7_conv(x)
        return y, vq_loss, util, {}     # empty aux
    elif self.decoder_type == "v6_fc":
        return self._forward_v6_fc(x)   # returns 4-tuple with aux
    elif self.decoder_type == "tf2_128":
        return self._forward_tf2_128(x) # returns 4-tuple with aux

Training loop 호출부는 unpack 을 y, vq_loss, util, aux = model(x) 로 갱신. 기존 3-tuple unpack 사용처가 있다면 호환 shim 추가.

3.3b Training loss 변경 (critic F3 보강 2 — loss_decode)¶

v7_runner training loop 의 loss 계산에 loss_decode 추가 (W1/W2 에서만 활성). v6 R1b line 408 pattern 복원:

# training loop within v7_runner.py (main training function)
y_hat, vq_loss, util, aux = model(x)

# peak-aware primary loss
loss_main = peak_weighted_smooth_l1(y_hat, y_true, alpha=2.0, beta=0.1)

# critic F3 보강 2: W1/W2 에서 auxiliary decoder supervision
if aux and "y_decode_intermediate" in aux:
    # target_norm 은 y_true 를 x normalize 와 동일 방식으로 정규화
    y_true_sq = y_true.squeeze(-1)
    y_mean = aux.get("x_mean") if "x_mean" in aux else aux["x_norm"].mean(dim=1, keepdim=True)
    # 단순화: x normalization stats 를 사용 (v6 R1b 와 동일 가정; 필요 시 별도 norm)
    # 정확한 v6 동작을 위해 aux 에 mean/std 도 포함하여 전달 권장
    loss_decode = F.smooth_l1_loss(
        aux["y_decode_intermediate"],
        y_true_sq_normalized,   # engineer 가 training loop 에서 계산
    )
else:
    loss_decode = 0.0

loss = loss_main + vq_loss + loss_decode

주의: engineer 는 v6 R1b v6_0415_fedpm_original.py:408 을 직접 참조하여 loss_decode target 이 어떤 normalize stats 를 쓰는지 확인할 것. aux dict 에 x_mean, x_std 를 추가 반환하는 것이 가장 안전.

3.3c Optimizer 변경 (critic F3 보강 3 — DLinear differential LR)¶

v6 R1b line 308-313 pattern 복원. v7_runner 의 optimizer 생성 코드에 CellSpec.dlinear_lr_scale 적용:

# v7_runner.py, optimizer 생성 위치 (training function 내)
dlinear_params = list(model.dlinear.parameters()) if model.dlinear is not None else []
dlinear_param_ids = {id(p) for p in dlinear_params}
other_params = [p for p in model.parameters() if id(p) not in dlinear_param_ids]

param_groups = [{"params": other_params, "lr": base_lr}]
if dlinear_params:
    param_groups.append({
        "params": dlinear_params,
        "lr": base_lr * cell_spec.dlinear_lr_scale,   # 1.0 default, 0.1 for W1/W2
    })

optimizer = torch.optim.AdamW(param_groups, weight_decay=weight_decay)

CellSpec dlinear_lr_scale = 1.0 (default, 기존 cell) → 단일 group 과 동일 결과 (backward-compat 유지)
CellSpec dlinear_lr_scale = 0.1 (W1, W2) → DLinear group 이 0.1 × base_lr 로 학습

3.4 TransformerDecoderW2 클래스¶

v7_runner.py 내부 정의 (별도 모듈 추출은 scope 아님):

class TransformerDecoderW2(nn.Module):
    """W2: Transformer encoder (2-layer) as decoder with global output projection."""

    def __init__(
        self,
        d_model: int = 64,
        nhead: int = 4,
        d_hid: int = 128,
        nlayers: int = 2,
        seq_len: int = 24,
        pred_len: int = 24,
        dropout: float = 0.0,
    ) -> None:
        super().__init__()
        enc_layer = nn.TransformerEncoderLayer(
            d_model=d_model,
            nhead=nhead,
            dim_feedforward=d_hid,
            batch_first=True,
            dropout=dropout,
            norm_first=False,
        )
        self.tf = nn.TransformerEncoder(enc_layer, num_layers=nlayers)
        self.pos_embed = nn.Parameter(torch.zeros(1, seq_len, d_model))
        self.out_proj = nn.Linear(d_model * seq_len, pred_len)

    def forward(self, x: torch.Tensor) -> tuple[torch.Tensor, torch.Tensor]:
        # x: [B, seq_len=24, d_model=64]
        x = x + self.pos_embed
        x = self.tf(x)                 # [B, 24, 64]
        x_flat = x.flatten(1)          # [B, 1536]
        y = self.out_proj(x_flat)      # [B, 24]   final pred
        # critic F3 보강 2: intermediate representation for loss_decode auxiliary supervision
        # 단순 선택: mean-pool over time as intermediate scalar-per-step proxy
        # 또는 self.out_proj 를 제거하지 않되 intermediate 는 pre-pool representation의 last-dim mean
        y_decode_intermediate = x.mean(dim=-1)   # [B, 24]  — channel-avg as intermediate
        return y, y_decode_intermediate

주의: XcodeYtimeDecoder (v6 원본) 은 현재 y 1-tuple 만 반환. W1 에서 y_decode_intermediate 를 얻기 위해 XcodeYtimeDecoder 를 수정 금지 조항에 걸림. 해결: - Option A: XcodeYtimeDecoder 의 내부 linear_out 직전 hidden state 를 forward hook 으로 캡처 — v6 lib 수정 없이 가능 - Option B: W1 에서 XcodeYtimeDecoder 를 사용하되 _forward_v6_fc 내부에서 FCDecoderBackbone 까지만 호출하고 linear_out 을 별도로 노출 — v7_runner 쪽에서만 수정 - Option A 가 v6 lib 수정 금지 조항 준수 + 최소 침습. engineer 는 self.decoder._output_hidden = {} register_forward_hook 또는 직접 submodule 접근 사용.

3.5 CELL_REGISTRY 확장¶

v7_runner.py:183 의 CELL_REGISTRY 에 W1, W2 추가 (기존 12종 행동 불변):

CELL_REGISTRY: dict[str, CellSpec] = {
    # ... existing B0 ~ V5 ...
    "W1": CellSpec(
        name="v6 FC-decoder + CNN-residual encoder + VQ + DLinear + peak-loss",
        method="proposed",
        peak_loss=True,
        use_vq=True,
        use_dlinear=True,
        vq_alignment=False,
        vq_reset=False,
        decoder_type="v6_fc",
        dlinear_lr_scale=0.1,    # v6 R1b differential LR
    ),
    "W2": CellSpec(
        name="Transformer-2L decoder + v7 encoder + VQ + DLinear + peak-loss",
        method="proposed",
        peak_loss=True,
        use_vq=True,
        use_dlinear=True,
        vq_alignment=False,
        vq_reset=False,
        decoder_type="tf2_128",
        dlinear_lr_scale=0.1,    # v6 R1b differential LR
    ),
}

_ALLOWED_CELLS (line 116~119) 에도 "W1", "W2" 추가.

3.6 CLI 지원 확인¶

v7_runner.py CLI 에서 --cells=W1 이 허용되는지 확인 (_validate_cells 에서 CELL_REGISTRY 체크만 하므로 자동 지원될 것으로 예상).

3.7 MLflow 로깅 추가¶

ProposedModel 인스턴스화 직후:

mlflow.log_param("decoder_type", cell_spec.decoder_type)
decoder_params = sum(p.numel() for p in model.decoder.parameters())
mlflow.log_param("decoder_param_count", decoder_params)

4. Unit Tests (필수 5개 + α)¶

tests/test_track_f_decoder.py (신규 파일):

import pytest
import torch
from experiments.federated.v7_runner import ProposedModel, CELL_REGISTRY


def test_legacy_cells_backward_compat():
    """B0~V5 cell의 default decoder_type=v7_conv 및 param count 불변."""
    for cell_id in ["B0", "B1", "B2", "B3", "B4", "A1", "A2", "A3", "A4", "V1", "V2", "V3", "V4", "V5"]:
        spec = CELL_REGISTRY[cell_id]
        assert spec.decoder_type == "v7_conv", f"{cell_id} changed decoder_type"

    # ProposedModel default
    m = ProposedModel(use_dlinear=True)
    assert m.decoder_type == "v7_conv"
    # Conv1d x2 + pred_head = 24,769
    decoder_params = sum(p.numel() for p in m.decoder.parameters())
    pred_head_params = sum(p.numel() for p in m.pred_head.parameters())
    assert decoder_params + pred_head_params == 24_769


def test_w1_decoder_params():
    """W1 decoder_type=v6_fc의 decoder 파라미터 수."""
    m = ProposedModel(decoder_type="v6_fc", use_dlinear=True)
    n = sum(p.numel() for p in m.decoder.parameters())
    assert n == 956_696, f"W1 decoder param count mismatch: {n}"
    # pred_head는 None
    assert m.pred_head is None


def test_w1_encoder_params():
    """W1 v6-style CNN-residual encoder 파라미터 수."""
    m = ProposedModel(decoder_type="v6_fc", use_dlinear=True)
    n = sum(p.numel() for p in m.encoder.parameters())
    assert n == 156_288, f"W1 encoder param count mismatch: {n}"


def test_w1_forward_shape():
    """W1 forward pass 출력 shape."""
    m = ProposedModel(decoder_type="v6_fc", use_dlinear=True).eval()
    x = torch.randn(4, 96, 1)
    with torch.no_grad():
        y, vq_loss, util = m(x)
    assert y.shape == (4, 24, 1)
    assert vq_loss.ndim == 0
    assert 0.0 <= util <= 1.0


def test_w2_decoder_params():
    """W2 decoder_type=tf2_128의 decoder 파라미터 수."""
    m = ProposedModel(decoder_type="tf2_128", use_dlinear=True)
    n = sum(p.numel() for p in m.decoder.parameters())
    assert n == 105_368, f"W2 decoder param count mismatch: {n}"
    assert m.pred_head is None


def test_w2_forward_shape():
    """W2 forward pass 출력 shape."""
    m = ProposedModel(decoder_type="tf2_128", use_dlinear=True).eval()
    x = torch.randn(4, 96, 1)
    with torch.no_grad():
        y, vq_loss, util = m(x)
    assert y.shape == (4, 24, 1)


def test_unknown_decoder_type_raises():
    """알 수 없는 decoder_type은 ValueError."""
    with pytest.raises(ValueError):
        ProposedModel(decoder_type="unknown_type", use_dlinear=True)


def test_w1_vq_state_still_shared():
    """W1이 여전히 VQ codebook만 FL 공유 (get_vq_state 호출 가능)."""
    m = ProposedModel(decoder_type="v6_fc", use_dlinear=True)
    state = m.get_vq_state()
    # VQ codebook size = 256 * 64 = 16384 elements in one entry
    assert any("codebook" in k or "embed" in k for k in state.keys())


# --- Critic F3 보강 3건 검증 ---

def test_w1_revin_biased_std():
    """W1 RevIN 이 biased std (unbiased=False, N 분모) 사용 — v6 R1b 일치."""
    m = ProposedModel(decoder_type="v6_fc", use_dlinear=True).eval()
    x = torch.randn(4, 96, 1)
    # biased std 기대값 수동 계산
    x_sq = x.squeeze(-1)
    mean_expected = x_sq.mean(dim=1, keepdim=True)
    var_biased = x_sq.var(dim=1, keepdim=True, unbiased=False)
    std_expected = var_biased.sqrt() + 1e-8
    # forward 내부 normalize 값이 동일해야 함
    # (engineer 는 aux 에 "mean" "std" 를 노출하거나 hook 으로 검증 가능하도록 구현)
    with torch.no_grad():
        _, _, _, aux = m(x)
    assert "x_norm" in aux
    # x_norm = (x - mean) / std, mean/std 는 biased 여야 함
    # 재구성: x_norm * std + mean ≈ x
    recon = aux["x_norm"] * std_expected + mean_expected
    assert torch.allclose(recon, x_sq, atol=1e-5)


def test_w1_aux_returns_decode_intermediate():
    """W1 forward 가 aux['y_decode_intermediate'] 반환 — loss_decode 용."""
    m = ProposedModel(decoder_type="v6_fc", use_dlinear=True).eval()
    x = torch.randn(4, 96, 1)
    with torch.no_grad():
        y, vq_loss, util, aux = m(x)
    assert "y_decode_intermediate" in aux
    assert aux["y_decode_intermediate"].shape == (4, 24)


def test_w1_dlinear_lr_scale_spec():
    """W1 CellSpec.dlinear_lr_scale=0.1 검증."""
    from experiments.federated.v7_runner import CELL_REGISTRY
    assert CELL_REGISTRY["W1"].dlinear_lr_scale == 0.1
    assert CELL_REGISTRY["W2"].dlinear_lr_scale == 0.1
    # 기존 cell 은 default 1.0 유지
    for cell_id in ["B0", "A3", "V5"]:
        assert CELL_REGISTRY[cell_id].dlinear_lr_scale == 1.0


def test_legacy_forward_shape_extended():
    """기존 v7_conv 경로 forward 가 4-tuple (y, vq_loss, util, aux) 반환하되 aux 는 empty."""
    m = ProposedModel(decoder_type="v7_conv", use_dlinear=True).eval()
    x = torch.randn(4, 96, 1)
    with torch.no_grad():
        result = m(x)
    assert len(result) == 4
    y, vq_loss, util, aux = result
    assert y.shape == (4, 24, 1)
    assert aux == {} or "y_decode_intermediate" not in aux

4.1 Regression 확인¶

기존 테스트 전체 pass 확인:

uv run python -m pytest tests/ -v --ignore=tests/integration_distilts.py

기대: 155+ tests + 신규 7개 모두 pass.

5. Acceptance Criteria (engineer 완료 선언 전 반드시 확인)¶

6. 리스크 완화 가이드¶

R1. v6 `lib/` import 실패¶

증상: ModuleNotFoundError: No module named 'lib.models.Encoder_and_Retrieval'

처리: src/FedUnit-64D1/lib/ 를 sys.path에 추가 (기존 v6_0415_fedpm_original.py:36-38 참조하여 동일 방식 사용):

_FEDUNIT_PATH = Path(__file__).resolve().parents[2] / "src/FedUnit-64D1"
if str(_FEDUNIT_PATH) not in sys.path:
    sys.path.insert(0, str(_FEDUNIT_PATH))

주의: 이것을 v7_runner.py 상단에 추가하는 것은 default behavior 변경이 되므로, _forward_v6_fc 내부 import 직전에 조건부 추가 권장.

R2. Encoder output shape 불일치¶

증상: v6 Encoder(...) output [B, D, seq] 가 v7 VectorQuantizer [B, L, D] 기대와 불일치
처리: forward pass에서 z = z.transpose(1, 2) 명시적 적용 (코드 예시에 반영됨)

R3. XcodeYtimeDecoder batch_first 문제¶

증상: XcodeYtimeDecoder 는 [seq, B, d_in] (batch_first=False) 기대 — v7 convention과 다름
처리: forward pass에서 z_hat.transpose(0, 1) 로 축 순서 맞춤

R4. RevIN vs in-function normalization 불일치 (critic F3 보강 1)¶

v6 R1b는 RevIN(num_features=1, affine=False) 사용 + biased std (unbiased=False, N 분모)
v7 ProposedModel 은 (x - mean) / std 인라인 계산, std 는 PyTorch default unbiased=True (N-1 분모) → v6 와 불일치
해결 (mandatory): W1/W2 inline normalization 에서 torch.var(x, dim=1, keepdim=True, unbiased=False).sqrt() + eps 사용. test_w1_revin_biased_std 로 검증.
옵션: lib/models/revin.py 의 RevIN(affine=False) 을 직접 import 도 허용 (v6 lib 수정 금지 조항 준수).

R5. forward pass의 pred_head=None 케이스¶

W1, W2 둘 다 pred_head 없음
테스트/체크포인트 로드 로직에서 if self.pred_head is not None: 가드 필요한지 점검

7. 보고 요구사항¶

engineer 완료 선언 시 다음 정보 반드시 포함: 1. 추가한 파일 경로 (모든 신규 파일) 2. 수정된 파일 라인 범위 3. 실행한 테스트 명령 + 출력 요약 (pass/fail 카운트) 4. 새 테스트 7개 개별 PASS 확인 5. ProposedModel(decoder_type="v7_conv") default param count가 변화 없음을 증명하는 snapshot 6. W1 + W2 forward smoke (random 입력 1회) 결과 shape 확인 출력

8. 금지된 우회¶

"테스트가 너무 엄격하니 assertion 제거" — 절대 금지
"import 경로 꼬이니 lib/ 복사" — 절대 금지 (import 사용)
"default decoder_type을 None으로 바꿔서 기존 코드 호환" — 금지 (default 반드시 "v7_conv" 문자열)
"unit test에서 param count tolerance 추가" — tolerance 0. 정확히 956,696 / 105,368 / 24,769.
"commit 할 때 verification hook skip" — 금지.

9. 이후 절차¶

engineer 완료 후 → exp-expert 가 track-f.1 smoke 실행 → Gate 판정 → (PASS 시) track-f.2 → track-f.3 → exp-critic 적대적 검토 → reporter.

10. 관련 문서¶

ADR: docs/decisions/ADR-009_v8_to_track_f_decoder_swap.md
Design spec: docs/reference/project_state/track_f_decoder_swap.md
TODO: todos/track-f_decoder_swap.md
심층 분석: report/version8/exp-expert/track_f_decoder_analysis.md