fix(deploy): 기존 runtime 쌍의 조건부 복구 검증

.github/workflows/deploy.yml

@@ -66,7 +66,4 @@ jobs:
       - name: Apply to AKS
         run: |
           trap 'rm -f "$KUBECONFIG"' EXIT
-          kubectl apply -f "$RUNNER_TEMP/release-manifests/backend-deployment.yaml" -n naruon-dev
-          kubectl rollout status deployment/backend -n naruon-dev --timeout=120s
-          kubectl apply -f "$RUNNER_TEMP/release-manifests/frontend-deployment.yaml" -n naruon-dev
-          kubectl rollout status deployment/frontend -n naruon-dev --timeout=120s
+          bash scripts/deploy_runtime_manifests.sh "$RUNNER_TEMP/release-manifests"

AGENTS.md

@@ -110,6 +110,21 @@ in this repo.
 
 ## Release governance defaults
 
+- CHANGELOG나 저장소 요약이 endpoint를 구현·배포됐다고 설명해도 현재 라우트와
+  실제 응답을 확인한다. 과거 branch의 코드를 찾으면 선행 PR과 유효 delta를 추적하고,
+  문서에 적힌 `/readyz`나 root 200을 DB readiness 근거로 승계하지 않는다.
+- 부분 배포 복구는 이전 image만 바꾸는 작업이 아니다. 같은 실행에서 `umask 077`로
+  이전 spec과 실제 적용 응답·readback을 확보하고 UID·spec 소유권을 확인한다.
+  정방향 변경과 복구 모두 UID·resourceVersion 조건부 patch를 사용한다. 다른 writer의
+  변경, 재생성, 충돌, 응답 유실은 강제 복구나 재시도 없이 실패로 남긴다.
+  서버 dry-run의 last-applied annotation을 그대로 저장하지 말고 spec만 변경한다.
+  복구 후 rollout과 최종 UID·spec readback까지 확인해도 원래 배포 job은 실패다.
+  두 객체의 복구가 끝난 뒤 양쪽 UID·spec을 다시 확인한다. 개별 복구 성공을
+  전체 복구 성공으로 합산하지 않으며 두 객체의 복구를 원자적이라고 표현하지 않는다.
+  사전 검사는 두 대상 모두 끝내고 나서 쓴다. 두 번째 대상의 누락·tag 기반 이전
+  상태·metadata 변경을 주입해 첫 번째 대상에도 patch가 없었는지 검사한다.
+  원시 snapshot·kubeconfig는 로그나 artifact로 올리지 않는다. 실제 workflow 호출의
+  부분 실패와 cleanup도 테스트하며 unit double을 실제 클러스터 검증으로 보고하지 않는다.
 - 배포 이미지의 tag는 게시 artifact의 동일성을 증명하지 않는다. matrix별 digest를
   같은 실행·revision·attempt의 개별 artifact로 전달하고 두 runtime의 sha256을
   모두 검증한 뒤 `image@sha256` manifest를 생성한다. 누락·변조·잘못된 digest는

backend/tests/fixtures/runtime_restore_kubectl.py

@@ -0,0 +1,83 @@
+"""Unit-only kubectl double; never opens a network connection."""
+
+import copy
+import json
+import os
+from pathlib import Path
+import sys
+
+state_path = Path(os.environ["RESTORE_TEST_STATE"])
+test_state = json.loads(state_path.read_text())
+command_args = sys.argv[1:]
+operation_name = command_args[0]
+test_state["calls"].append(operation_name)
+image_component = "backend"
+desired_state = None
+if "resources" in test_state:
+    if operation_name == "apply":
+        file_flag = "-f" if "-f" in command_args else "--filename"
+        desired_state = json.loads(Path(command_args[command_args.index(file_flag) + 1]).read_text())
+        image_component = desired_state["metadata"]["name"]
+    else:
+        image_component = command_args[2].removeprefix("deployment/")
+    test_state["current"] = test_state["resources"][image_component]
+    test_state["operations"].append([operation_name, image_component])
+exit_code = 0
+if operation_name == "get":
+    if test_state["scenario"] == "missing_resource" or (test_state["scenario"] == "frontend_missing" and image_component == "frontend"):
+        exit_code = 1
+    else:
+        print(json.dumps(test_state["current"]))
+elif operation_name == "apply" and "--dry-run=server" in command_args:
+    proposed_state = copy.deepcopy(test_state["current"])
+    proposed_state["metadata"].update(desired_state["metadata"])
+    proposed_state["spec"] = desired_state["spec"]
+    proposed_state["spec"]["revisionHistoryLimit"] = 10
+    proposed_state["metadata"].setdefault("annotations", {})["kubectl.kubernetes.io/last-applied-configuration"] = "unit-dry-run-annotation"
+    print(json.dumps(proposed_state))
+elif operation_name == "patch":
+    patch_path = Path(command_args[command_args.index("--patch-file") + 1])
+    patch_ops = json.loads(patch_path.read_text())
+    test_state["patch_ops"] = patch_ops
+    if test_state["scenario"] == "conflict":
+        test_state["current"]["metadata"]["resourceVersion"] = "external-update"
+    for patch_op in patch_ops:
+        if patch_op["op"] == "test":
+            current_value = test_state["current"]
+            for path_part in patch_op["path"].strip("/").split("/"):
+                current_value = current_value[path_part]
+            if current_value != patch_op["value"]:
+                exit_code = 1
+    if not exit_code:
+        replacement = next(item["value"] for item in patch_ops if item["op"] == "replace" and item["path"] == "/spec")
+        test_state["current"]["spec"] = replacement
+        test_state["current"]["metadata"]["resourceVersion"] = "restored-version"
+        forward_patch = False
+        if "resources" in test_state:
+            forward_patch = image_component not in test_state["applied_components"]
+            state_key = "applied_components" if forward_patch else "restored_components"
+            test_state[state_key].append(image_component)
+        if test_state["scenario"] == "response_lost" or (forward_patch and image_component == "frontend" and test_state["scenario"] == "frontend_response_lost"):
+            exit_code = 1
+        else:
+            print(json.dumps(test_state["current"]))
+elif operation_name == "rollout":
+    if (
+        "resources" in test_state and image_component == "frontend"
+        and image_component in test_state["applied_components"]
+        and image_component not in test_state["restored_components"]
+        and test_state["scenario"] in {"frontend_rollout_failed", "backend_drift", "restored_frontend_drift"}
+    ):
+        exit_code = 1
+        if test_state["scenario"] == "backend_drift":
+            test_state["resources"]["backend"]["spec"]["replicas"] = 99
+    if test_state["scenario"] == "unhealthy_restore":
+        exit_code = 1
+    if test_state["scenario"] == "readback_drift":
+        test_state["current"]["spec"]["replicas"] = 99
+    if test_state["scenario"] == "restored_frontend_drift" and image_component == "backend" and image_component in test_state["restored_components"]:
+        test_state["resources"]["frontend"]["spec"]["replicas"] = 99
+else:
+    exit_code = 2
+state_path.write_text(json.dumps(test_state))
+raise SystemExit(exit_code)

backend/tests/test_release_manifest_digests.py

@@ -111,8 +111,7 @@ def test_release_workflow_passes_separate_same_revision_artifacts() -> None:
     render = next(step for step in steps if step.get("id") == "render_manifests")
     assert "scripts/render_release_manifests.sh" in render["run"]
     apply_step = next(step for step in steps if step["name"] == "Apply to AKS")
-    assert '"$RUNNER_TEMP/release-manifests/backend-deployment.yaml"' in apply_step["run"]
-    assert '"$RUNNER_TEMP/release-manifests/frontend-deployment.yaml"' in apply_step["run"]
+    assert 'bash scripts/deploy_runtime_manifests.sh "$RUNNER_TEMP/release-manifests"' in apply_step["run"]
 
 
 @pytest.mark.parametrize("artifact_state", ["valid", "missing_backend", "missing_frontend", "invalid_backend", "invalid_frontend"])

backend/tests/test_runtime_deployment_restore.py

@@ -0,0 +1,144 @@
+"""Run conditional restoration against a network-free kubectl unit double."""
+
+import copy
+import json
+import os
+from pathlib import Path
+import shlex
+import shutil
+import subprocess
+import sys
+
+import pytest
+import yaml
+
+
+REPO_ROOT = Path(__file__).resolve().parents[2]
+
+
+@pytest.mark.parametrize("scenario", ["restored", "uid_changed", "spec_changed", "conflict", "response_lost", "unhealthy_restore", "readback_drift", "missing_resource"])
+def test_restore_uses_owned_spec_and_atomic_resource_version(
+    tmp_path: Path, scenario: str
+) -> None:
+    """Never overwrite a recreated/drifted resource or report ambiguous success."""
+    before_state = {
+        "kind": "Deployment",
+        "metadata": {"name": "backend", "namespace": "naruon-dev", "uid": "unit-backend", "resourceVersion": "before-version"},
+        "spec": {"replicas": 2, "template": {"spec": {"containers": [{"name": "backend", "image": "ghcr.io/unit/backend@sha256:" + "a" * 64}]}}},
+    }
+    owned_state = copy.deepcopy(before_state)
+    owned_state["metadata"]["resourceVersion"] = "applied-version"
+    owned_state["spec"]["replicas"] = 3
+    owned_state["spec"]["template"]["spec"]["containers"][0]["image"] = "ghcr.io/unit/backend@sha256:" + "b" * 64
+    current_state = copy.deepcopy(owned_state)
+    current_state["metadata"]["resourceVersion"] = "controller-status-update"
+    if scenario == "uid_changed":
+        current_state["metadata"]["uid"] = "replacement-resource"
+    if scenario == "spec_changed":
+        current_state["spec"]["replicas"] = 7
+    for file_name, payload in (("before.json", before_state), ("owned.json", owned_state)):
+        (tmp_path / file_name).write_text(json.dumps(payload))
+    state_path = tmp_path / "unit_state.json"
+    state_path.write_text(json.dumps({"scenario": scenario, "current": current_state, "calls": []}))
+    fake_binary = tmp_path / "kubectl"
+    fixture_path = REPO_ROOT / "backend/tests/fixtures/runtime_restore_kubectl.py"
+    fake_binary.write_text(f"#!/bin/sh\nexec {shlex.quote(sys.executable)} {shlex.quote(str(fixture_path))} \"$@\"\n")
+    fake_binary.chmod(0o700)
+    jq_binary = shutil.which("jq")
+    assert jq_binary, "jq is required for the deployment contract"
+    result = subprocess.run(
+        ["bash", str(REPO_ROOT / "scripts/restore_runtime_deployment.sh"), "backend", str(tmp_path / "before.json"), str(tmp_path / "owned.json")],
+        env={"PATH": os.pathsep.join([str(tmp_path), str(Path(jq_binary).parent), os.defpath]), "RUNNER_TEMP": str(tmp_path), "RESTORE_TEST_STATE": str(state_path)},
+        capture_output=True, text=True, check=False,
+    )
+    final_state = json.loads(state_path.read_text())
+    if scenario == "restored":
+        assert result.returncode == 0, result.stderr
+        assert final_state["current"]["spec"] == before_state["spec"]
+        assert final_state["calls"] == ["get", "patch", "rollout", "get"]
+        assert {item["path"] for item in final_state["patch_ops"] if item["op"] == "test"} >= {"/metadata/uid", "/metadata/resourceVersion"}
+        assert "restore_verified:backend" in result.stdout
+    else:
+        assert result.returncode != 0
+        assert "restore_verified:" not in result.stdout
+        assert "restore_failed:" in result.stderr
+        if scenario in {"uid_changed", "spec_changed", "missing_resource"}:
+            assert "patch" not in final_state["calls"]
+
+
+@pytest.mark.parametrize("scenario", ["deployed", "frontend_rollout_failed", "backend_drift", "frontend_response_lost", "restored_frontend_drift", "frontend_tag_baseline", "frontend_metadata_change", "frontend_missing"])
+def test_partial_deployment_restores_only_confirmed_owned_resources(
+    tmp_path: Path, scenario: str
+) -> None:
+    """Run the real caller and restore helper with admission-normalized objects."""
+    resource_states = {}
+    manifest_directory = tmp_path / "release-manifests"
+    manifest_directory.mkdir()
+    for image_component in ("backend", "frontend"):
+        resource_states[image_component] = {
+            "kind": "Deployment",
+            "metadata": {"name": image_component, "namespace": "naruon-dev", "uid": "unit-" + image_component, "resourceVersion": "old-version"},
+            "spec": {"replicas": 2, "revisionHistoryLimit": 10, "template": {"spec": {"containers": [{"name": image_component, "image": "ghcr.io/unit/" + image_component + "@sha256:" + "a" * 64}]}}},
+        }
+        if image_component == "frontend" and scenario == "frontend_tag_baseline":
+            resource_states[image_component]["spec"]["template"]["spec"]["containers"][0]["image"] = "ghcr.io/unit/frontend:old-version"
+        desired_state = copy.deepcopy(resource_states[image_component])
+        if image_component == "frontend" and scenario == "frontend_metadata_change":
+            desired_state["metadata"]["labels"] = {"release_owner": "changed-owner"}
+        desired_state["spec"]["replicas"] = 3
+        del desired_state["spec"]["revisionHistoryLimit"]
+        desired_state["spec"]["template"]["spec"]["containers"][0]["image"] = "ghcr.io/unit/" + image_component + "@sha256:" + "b" * 64
+        (manifest_directory / f"{image_component}-deployment.yaml").write_text(json.dumps(desired_state))
+    prior_states = copy.deepcopy(resource_states)
+    state_path = tmp_path / "unit_state.json"
+    state_path.write_text(json.dumps({"scenario": scenario, "resources": resource_states, "calls": [], "operations": [], "applied_components": [], "restored_components": []}))
+    fake_binary = tmp_path / "kubectl"
+    fixture_path = REPO_ROOT / "backend/tests/fixtures/runtime_restore_kubectl.py"
+    fake_binary.write_text(f"#!/bin/sh\nexec {shlex.quote(sys.executable)} {shlex.quote(str(fixture_path))} \"$@\"\n")
+    fake_binary.chmod(0o700)
+    jq_binary = shutil.which("jq")
+    assert jq_binary
+    kubeconfig_path = tmp_path / "unit_kubeconfig"
+    kubeconfig_path.write_text("unit-only-not-a-credential\n")
+    workflow = yaml.safe_load((REPO_ROOT / ".github/workflows/deploy.yml").read_text())
+    apply_command = next(step["run"] for step in workflow["jobs"]["deploy"]["steps"] if step["name"] == "Apply to AKS")
+    result = subprocess.run(
+        ["bash", "-euo", "pipefail", "-c", apply_command],
+        cwd=REPO_ROOT,
+        env={"PATH": os.pathsep.join([str(tmp_path), str(Path(jq_binary).parent), os.defpath]), "RUNNER_TEMP": str(tmp_path), "RESTORE_TEST_STATE": str(state_path), "KUBECONFIG": str(kubeconfig_path)},
+        capture_output=True, text=True, check=False,
+    )
+    final_state = json.loads(state_path.read_text())
+    assert not kubeconfig_path.exists()
+    if scenario in {"frontend_tag_baseline", "frontend_metadata_change", "frontend_missing"}:
+        assert result.returncode != 0
+        assert final_state["applied_components"] == []
+        assert final_state["restored_components"] == []
+        assert "patch" not in final_state["calls"]
+        assert final_state["resources"] == prior_states
+        assert "verified" not in result.stdout + result.stderr
+        return
+    assert final_state["applied_components"] == ["backend", "frontend"]
+    for image_component in ("backend", "frontend"):
+        assert final_state["resources"][image_component]["metadata"].get("annotations", {}) == prior_states[image_component]["metadata"].get("annotations", {})
+    if scenario == "deployed":
+        assert result.returncode == 0, result.stderr
+        assert final_state["restored_components"] == []
+        assert "deployment_verified" in result.stdout
+    elif scenario == "frontend_rollout_failed":
+        assert result.returncode != 0
+        assert final_state["restored_components"] == ["frontend", "backend"]
+        for image_component in ("backend", "frontend"):
+            assert final_state["resources"][image_component]["spec"] == prior_states[image_component]["spec"]
+        assert "rollback_verified" in result.stderr
+    elif scenario == "restored_frontend_drift":
+        assert result.returncode != 0
+        assert final_state["restored_components"] == ["frontend", "backend"]
+        assert final_state["resources"]["frontend"]["spec"]["replicas"] == 99
+        assert "rollback_verified" not in result.stderr
+    else:
+        assert result.returncode != 0
+        assert final_state["restored_components"] == []
+        assert "rollback_verified" not in result.stderr
+        if scenario == "backend_drift":
+            assert final_state["resources"]["backend"]["spec"]["replicas"] == 99