fix(studio-save): 응답 껍데기를 벗기는 자리를 코드가 갖고, 계획마다 새 키가 나게 한다

① GET /documents/{id} 는 문서를 data.document 아래에 둔다(studio-v1.yaml:1831). A 가
   B-011 에서 data 를 문서로 보고 견줘 칸이 전부 「다르다」로 나왔다 — 벗기는 자리를
   코드가 갖고 있지 않아서 난 일이라 unwrap_document 를 둔다. 모양이 다르면 조용히
   넘기지 않고 거절한다. 잘못된 층을 견주는 것보다 멈추는 편이 낫다.

   그리고 updatedAt 을 서버가 붙이는 칸으로 옮겼다. 매 저장마다 「서버가 더 줬다」로
   나오면 그 칸을 아무도 안 읽게 된다. variantIds 는 안 옮겼다 — 계약이 입력으로 받는
   칸인데 이 어댑터가 안 보내므로 보여야 한다.

② CR-007 때문에 같은 키를 두 번 못 쓴다. 런 식별자를 계획 파일에 적고 키에 섞는다.
   같은 계획을 두 번 보내면 같은 키, 계획을 새로 만들면 다른 키다. 보낼 때 만들면
   재시도가 곧 중복 생성이 되므로 만드는 자리는 계획을 짜는 순간 하나뿐이다.
   --run-id 를 안 주면 예전 키 모양 그대로라 옛 계획을 안 깬다.

python3 -m unittest discover -s scripts/tests — Ran 291 · OK (skipped=13)
PIPELINE CONTRACT: PASS

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01Q4vKjQo9KKBBokzxqXLCfk
This commit is contained in:
DongHyeonka
2026-09-11 13:18:30 +09:00
co-authored by Claude Opus 5
parent 1be1c8ea6d
commit d473609e0a
2 changed files with 148 additions and 7 deletions
+68 -7
View File
@@ -181,7 +181,8 @@ def _sha256_text(text: str) -> str:
return hashlib.sha256(text.encode("utf-8")).hexdigest() return hashlib.sha256(text.encode("utf-8")).hexdigest()
def idempotency_key(op: str, record_rel: str, payload: dict | None = None) -> str: def idempotency_key(op: str, record_rel: str, payload: dict | None = None,
run_id: str | None = None) -> str:
"""기록 경로에서 유도한 안정 키. """기록 경로에서 유도한 안정 키.
프런트는 호출마다 `crypto.randomUUID()` 로 새 키를 만든다(`local-id.ts:8`). 그래서 프런트는 호출마다 `crypto.randomUUID()` 로 새 키를 만든다(`local-id.ts:8`). 그래서
@@ -206,9 +207,19 @@ def idempotency_key(op: str, record_rel: str, payload: dict | None = None) -> st
키가 언제 사라지는지는 `IDEMPOTENCY_TTL_CONTRACT` 를 본다 — **운영 인스턴스의 실제 키가 언제 사라지는지는 `IDEMPOTENCY_TTL_CONTRACT` 를 본다 — **운영 인스턴스의 실제
값은 미확인이다.** 값은 미확인이다.**
**`run_id` 가 그 막힘을 푼다.** 런 식별자를 **계획 파일에 적고** 키에 섞는다.
- 같은 계획을 두 번 보내면 **같은 키**다. 계획 안에 적힌 값이라 보낼 때마다 안 바뀐다
- 계획을 새로 만들면 **다른 키**다. 그래서 지웠던 기록을 다시 걸 수 있다
**보낼 때 키를 만들어 내지 않는다.** 그러면 재시도가 곧 중복 생성이 된다 — 이 어댑터가
처음부터 막으려던 것이 그것이다. 만드는 자리는 계획을 짜는 순간 하나뿐이다.
""" """
base = _sha256_text(record_rel)[:32] base = _sha256_text(record_rel)[:32]
key = f"studio-{op}-{base}" key = f"studio-{op}-{base}"
if run_id:
key += "-" + _sha256_text(run_id)[:12]
if payload is not None: if payload is not None:
key += "-" + _sha256_text(json.dumps(payload, ensure_ascii=False, sort_keys=True))[:32] key += "-" + _sha256_text(json.dumps(payload, ensure_ascii=False, sort_keys=True))[:32]
# **지금 키 정책에서는 여기 안 온다.** 키가 `studio-{op}-{32자}`(+`-{32자}`)라 길이가 # **지금 키 정책에서는 여기 안 온다.** 키가 `studio-{op}-{32자}`(+`-{32자}`)라 길이가
@@ -773,7 +784,7 @@ def _reject_forbidden_paths(plan: list[dict]) -> None:
def plan_requests(record_path: str, doc: dict, document_id: str | None, def plan_requests(record_path: str, doc: dict, document_id: str | None,
expected_version: int | None, expected_version: int | None,
harness_test: bool = False) -> list[dict]: harness_test: bool = False, run_id: str | None = None) -> list[dict]:
"""보낼 요청을 그대로 적어 낸다. 보내지 않는다.""" """보낼 요청을 그대로 적어 낸다. 보내지 않는다."""
rel = os.path.relpath(os.path.abspath(record_path), ROOT) rel = os.path.relpath(os.path.abspath(record_path), ROOT)
steps: list[dict] = [] steps: list[dict] = []
@@ -793,7 +804,7 @@ def plan_requests(record_path: str, doc: dict, document_id: str | None,
if not document_id: if not document_id:
steps.append({ steps.append({
"op": "create", "method": "POST", "path": f"{BASE_PATH}/documents", "op": "create", "method": "POST", "path": f"{BASE_PATH}/documents",
"headers": {IDEMPOTENCY_HEADER: idempotency_key("create", rel), "headers": {IDEMPOTENCY_HEADER: idempotency_key("create", rel, None, run_id),
"X-CSRF-TOKEN": "<env STUDIO_CSRF_TOKEN>"}, "X-CSRF-TOKEN": "<env STUDIO_CSRF_TOKEN>"},
"cookies": {"TECHLOG_SESSION": "<env STUDIO_SESSION_COOKIE>"}, "cookies": {"TECHLOG_SESSION": "<env STUDIO_SESSION_COOKIE>"},
"body": doc, "body": doc,
@@ -815,7 +826,7 @@ def plan_requests(record_path: str, doc: dict, document_id: str | None,
body = {"expectedVersion": expected_version, "document": doc} body = {"expectedVersion": expected_version, "document": doc}
steps.append({ steps.append({
"op": "save", "method": "PUT", "path": f"{BASE_PATH}/documents/{document_id}", "op": "save", "method": "PUT", "path": f"{BASE_PATH}/documents/{document_id}",
"headers": {IDEMPOTENCY_HEADER: idempotency_key("save", rel, body), "headers": {IDEMPOTENCY_HEADER: idempotency_key("save", rel, body, run_id),
"X-CSRF-TOKEN": "<env STUDIO_CSRF_TOKEN>"}, "X-CSRF-TOKEN": "<env STUDIO_CSRF_TOKEN>"},
"cookies": {"TECHLOG_SESSION": "<env STUDIO_SESSION_COOKIE>"}, "cookies": {"TECHLOG_SESSION": "<env STUDIO_SESSION_COOKIE>"},
"body": body, "body": body,
@@ -837,7 +848,8 @@ def plan_requests(record_path: str, doc: dict, document_id: str | None,
"expect": {"status": 200, "readHeader": REPLAYED_HEADER, "expect": {"status": 200, "readHeader": REPLAYED_HEADER,
"onConflict": _expect_error("VERSION_CONFLICT"), "onConflict": _expect_error("VERSION_CONFLICT"),
"note": "멱등 키는 보낼 내용이 정해진 뒤에 만든다. " "note": "멱등 키는 보낼 내용이 정해진 뒤에 만든다. "
"scripts/studio-save.py 의 idempotency_key('save', 경로, 본문) 이다"}, "scripts/studio-save.py 의 "
"idempotency_key('save', 경로, 본문, runId) 이다"},
}) })
steps.append({ steps.append({
"op": "verify", "method": "GET", "op": "verify", "method": "GET",
@@ -946,6 +958,41 @@ ITEM_ID_UNMEASURED = ("이 종류는 서버가 항목 `id` 를 다시 매기는
"재서 ITEM_ID_REISSUED 에 적는다") "재서 ITEM_ID_REISSUED 에 적는다")
# `GET /documents/{id}` 의 응답은 문서를 **`data.document` 아래**에 둔다. `data` 에는
# `currentValidation`·`latestPreview`·`currentPublication`·`dependencyRevision`·`nextAction`
# 이 문서와 나란히 있다 (`studio-v1.yaml:1828-1847`).
DETAIL_SIBLINGS = ("currentValidation", "latestPreview", "currentPublication",
"dependencyRevision", "nextAction")
def unwrap_document(response: dict) -> tuple[dict, dict]:
"""운영 응답에서 **문서**와 **그 곁의 상태**를 갈라 낸다.
A 가 `B-011` 운영 실행에서 `data` 를 문서로 보고 견줘 **열세 칸이 전부 「다르다」로
나왔다.** 「전부 다르다」였기에 바로 틀린 줄 알았지, 한두 칸만 어긋났으면 결함으로
적고 넘어갔을 것이다. **껍데기를 벗기는 자리를 코드가 갖고 있지 않아서 난 일이라,
여기에 둔다.**
모양이 다르면 **조용히 넘기지 않고 거절한다.** 잘못된 층을 견주는 것보다 멈추는 편이
낫다 — 그 실수는 「전부 다르다」로 크게 나와 줄 때만 눈에 띈다.
"""
if not isinstance(response, dict):
raise Refused(f"응답이 객체가 아니다: {type(response).__name__}")
detail = response.get("data", response) # 봉투 없이 `data` 만 받은 경우도 받는다
if not isinstance(detail, dict):
raise Refused("응답의 data 가 객체가 아니다")
doc = detail.get("document")
if doc is None:
found = " · ".join(sorted(detail)[:8]) or "(빈 객체)"
raise Refused(
"응답에서 document 를 못 찾았다. `data.document` 아래에 문서가 있어야 한다"
f" (studio-v1.yaml:1831)\n 받은 칸: {found}\n"
" `data` 를 문서로 보고 견주면 모든 칸이 「다르다」로 나온다")
if not isinstance(doc, dict):
raise Refused(f"data.document 가 객체가 아니다: {type(doc).__name__}")
return doc, detail
def _is_empty(v) -> bool: def _is_empty(v) -> bool:
"""값이 「없음」인가. 서버가 빈 배열로, 어댑터가 `None` 으로 두는 자리를 같게 본다.""" """값이 「없음」인가. 서버가 빈 배열로, 어댑터가 `None` 으로 두는 자리를 같게 본다."""
return v is None or v == "" or v == [] or v == {} return v is None or v == "" or v == [] or v == {}
@@ -1033,8 +1080,12 @@ def compare_saved(sent: dict, fetched: dict) -> dict:
elif not raw_same: elif not raw_same:
whitespace_only.append({"field": field, whitespace_only.append({"field": field,
"note": "정규화 뒤에만 같다. 공백이나 줄바꿈이 달라졌다"}) "note": "정규화 뒤에만 같다. 공백이나 줄바꿈이 달라졌다"})
# 서버가 붙이는 칸. 어댑터가 안 보내는 것이 맞으므로 「서버가 더 줬다」로 세지 않는다.
# **`variantIds` 는 여기 없다** — 계약이 입력으로 받는 칸인데(`studio-v1.yaml:925`)
# 이 어댑터가 안 보낸다. 서버가 돌려주면 그것이 보여야 한다
SERVER_ASSIGNED = ("id", "version", "updatedAt")
unexpected = [k for k in fetched unexpected = [k for k in fetched
if k not in sent and k not in NOT_COMPARED and k not in ("id", "version")] if k not in sent and k not in NOT_COMPARED and k not in SERVER_ASSIGNED]
return { return {
"same": not differences and not whitespace_only, "same": not differences and not whitespace_only,
"differences": differences, "differences": differences,
@@ -1077,6 +1128,10 @@ def main() -> int:
ap.add_argument("--expected-version", type=int, help="GET 으로 읽은 현재 version") ap.add_argument("--expected-version", type=int, help="GET 으로 읽은 현재 version")
ap.add_argument("-o", "--out", help="계획을 적을 파일") ap.add_argument("-o", "--out", help="계획을 적을 파일")
ap.add_argument("--verdicts", help="경고마다 PASS/FAIL/UNKNOWN 을 적은 검토 판정 파일") ap.add_argument("--verdicts", help="경고마다 PASS/FAIL/UNKNOWN 을 적은 검토 판정 파일")
ap.add_argument("--run-id",
help="이 런의 식별자. 계획 파일에 적히고 멱등 키에 섞인다. 안 주면 "
"만든다 — 같은 계획을 두 번 보내면 같은 키, 계획을 새로 만들면 "
"다른 키다. 지웠던 기록을 다시 걸려면 새 계획이 필요하다(CR-007)")
ap.add_argument("--project-id", ap.add_argument("--project-id",
help="PROJECT_DECISION 이 걸리는 프로젝트의 uuid. 사람이 Studio 목록에서 " help="PROJECT_DECISION 이 걸리는 프로젝트의 uuid. 사람이 Studio 목록에서 "
"읽은 값이어야 한다 — 이 어댑터는 이름을 uuid 로 바꾸지 않는다. " "읽은 값이어야 한다 — 이 어댑터는 이름을 uuid 로 바꾸지 않는다. "
@@ -1109,8 +1164,11 @@ def main() -> int:
doc["projectId"] = args.project_id doc["projectId"] = args.project_id
if args.harness_test: if args.harness_test:
doc = as_harness_test(doc) doc = as_harness_test(doc)
# 안 주면 만든다. **계획 파일에 적히므로 그 뒤로는 안 바뀐다** — 보낼 때 만드는 것과
# 다르다. `CR-007` 때문에 같은 키를 두 번 못 쓰므로 새 계획은 새 키여야 한다
run_id = args.run_id or f"run-{uuid.uuid4()}"
steps = plan_requests(args.record, doc, args.document_id, args.expected_version, steps = plan_requests(args.record, doc, args.document_id, args.expected_version,
args.harness_test) args.harness_test, run_id)
if args.send and not UNATTENDED_SAVE_ENABLED: if args.send and not UNATTENDED_SAVE_ENABLED:
raise Refused("무인 저장이 꺼져 있다 — " + UNATTENDED_SAVE_BLOCKED_BY) raise Refused("무인 저장이 꺼져 있다 — " + UNATTENDED_SAVE_BLOCKED_BY)
@@ -1123,6 +1181,9 @@ def main() -> int:
"mode": "plan", "mode": "plan",
"unattendedSaveEnabled": UNATTENDED_SAVE_ENABLED, "unattendedSaveEnabled": UNATTENDED_SAVE_ENABLED,
"unattendedSaveBlockedBy": UNATTENDED_SAVE_BLOCKED_BY, "unattendedSaveBlockedBy": UNATTENDED_SAVE_BLOCKED_BY,
# **런 식별자는 계획 안에 있다.** 보낼 때 만들지 않는다 — 그러면 재시도가 곧
# 중복 생성이 된다. 같은 계획을 두 번 보내면 같은 키, 계획을 새로 만들면 다른 키다
"runId": run_id,
"record": os.path.relpath(os.path.abspath(args.record), ROOT), "record": os.path.relpath(os.path.abspath(args.record), ROOT),
"recordSha256": _sha256_file(args.record), "recordSha256": _sha256_file(args.record),
"packageSha256": _sha256_file(args.package), "packageSha256": _sha256_file(args.package),
+80
View File
@@ -7,10 +7,18 @@ import importlib.util
import json import json
import os import os
import subprocess import subprocess
import sys
import tempfile import tempfile
import unittest import unittest
ROOT = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) ROOT = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
# 런 식별자 회귀가 실제로 계획을 내 보려고 쓰는 입력. 관문이 전부 exit 0 인 프로젝트다
DECISION_RECORD = os.path.join(
"docs", "n+1liner", "tech-log-studio", "jpa-feed-query-performance", "decision",
"decision-keyset-for-feed-pagination.md")
PACKAGE = "/home/donghyeon/workspace/chat-gpt-container/success/reports/"\
"B-B011-plans/package.json"
PROJECT_ID = "8ff68a93-0cf6-4d91-a973-40e9f3d3e99d"
_spec = importlib.util.spec_from_file_location( _spec = importlib.util.spec_from_file_location(
"studio_save", os.path.join(ROOT, "scripts", "studio-save.py")) "studio_save", os.path.join(ROOT, "scripts", "studio-save.py"))
ss = importlib.util.module_from_spec(_spec) ss = importlib.util.module_from_spec(_spec)
@@ -902,6 +910,78 @@ class HarnessTestPlanTest(unittest.TestCase):
q = {"kind": "QUESTION", "facts": [{"id": "a", "text": "t", "order": 0}]} q = {"kind": "QUESTION", "facts": [{"id": "a", "text": "t", "order": 0}]}
self.assertNotIn("itemIdBehaviourUnmeasured", ss.compare_saved(q, dict(q))) self.assertNotIn("itemIdBehaviourUnmeasured", ss.compare_saved(q, dict(q)))
def test_the_envelope_is_unwrapped_before_comparing(self):
"""`GET /documents/{id}` 는 문서를 `data.document` 아래에 둔다
(`studio-v1.yaml:1831`). A 가 `data` 를 문서로 보고 견줘 칸이 전부 「다르다」로
나왔다 — **껍데기를 벗기는 자리를 코드가 갖고 있지 않아서 난 일이다.**"""
doc = {"kind": "PROJECT_DECISION", "title": "t", "statement": "s"}
env = {"success": True, "meta": {},
"data": {"document": doc, "currentPublication": None,
"currentValidation": None, "latestPreview": None,
"dependencyRevision": {}, "nextAction": "VALIDATE"}}
got, detail = ss.unwrap_document(env)
self.assertEqual(doc, got)
self.assertTrue(ss.is_private(detail)[0])
# 벗기면 같고, 안 벗기면 전부 다르다
self.assertTrue(ss.compare_saved(doc, got)["same"])
self.assertFalse(ss.compare_saved(doc, env["data"])["same"])
# `data` 만 줘도 받는다
self.assertEqual(doc, ss.unwrap_document(env["data"])[0])
def test_an_unexpected_envelope_is_refused_not_compared(self):
"""모양이 다르면 조용히 넘기지 않는다. 잘못된 층을 견주는 것보다 멈추는 편이 낫다 —
그 실수는 「전부 다르다」로 크게 나와 줄 때만 눈에 띈다."""
for bad in ({"success": True, "data": {"nextAction": "X"}},
{"success": True, "data": {"document": "문자열이다"}},
"객체가 아니다"):
with self.subTest(bad=str(bad)[:32]):
with self.assertRaises(ss.Refused):
ss.unwrap_document(bad)
def test_server_assigned_fields_are_not_reported_as_unexpected(self):
"""`updatedAt` 은 서버가 붙인다. 매 저장마다 「서버가 더 줬다」로 나오면
그 칸을 아무도 안 읽게 된다. **`variantIds` 는 다르다** — 계약이 입력으로 받는데
이 어댑터가 안 보내는 칸이라 보여야 한다."""
sent = {"kind": "CASE", "title": "t"}
got = {**sent, "id": "x", "version": 2, "updatedAt": "2026-09-11T00:00:00Z",
"variantIds": []}
self.assertEqual(["variantIds"], ss.compare_saved(sent, got)["unexpected"])
def test_a_new_plan_gets_a_new_key_but_one_plan_keeps_its_own(self):
"""`CR-007` 때문에 같은 키를 두 번 못 쓴다. 그래서 계획을 새로 만들면 새 키여야
한다 — 그런데 **보낼 때 만들면 재시도가 곧 중복 생성이 된다.**
런 식별자를 계획에 적어 그 둘을 함께 얻는다."""
rel = "docs/p/t/case/x.md"
a = ss.idempotency_key("create", rel, None, "run-aaa")
b = ss.idempotency_key("create", rel, None, "run-bbb")
self.assertNotEqual(a, b) # 계획이 새로 만들어지면 새 키
self.assertEqual(a, ss.idempotency_key("create", rel, None, "run-aaa"))
# 같은 계획을 두 번 보내는 것은 같은 값을 두 번 쓰는 것이다
self.assertLessEqual(len(a), ss.MAX_KEY_LENGTH)
# 런 식별자를 안 주면 예전 모양 그대로다 — 옛 계획을 깨지 않는다
self.assertEqual(ss.idempotency_key("create", rel),
ss.idempotency_key("create", rel, None, None))
def test_the_plan_records_its_run_id(self):
"""**보낼 때 만들지 않는다.** 계획 파일에 적혀 있어야 같은 계획이 같은 키를 쓴다."""
if not os.path.isfile(PACKAGE):
# 묶음은 worktree 밖에 있다. 없으면 **안 돌았다고 적고 넘어간다** —
# 조용히 통과시키면 이 회귀가 초록인 채로 아무것도 안 재게 된다
self.skipTest(f"묶음이 없어 계획을 못 낸다: {PACKAGE}")
with tempfile.TemporaryDirectory() as d:
out = os.path.join(d, "plan.json")
r = subprocess.run(
[sys.executable, "scripts/studio-save.py", "--record", DECISION_RECORD,
"--package", PACKAGE, "--harness-test", "--project-id", PROJECT_ID,
"--run-id", "run-fixed-for-the-test", "-o", out],
cwd=ROOT, capture_output=True, text=True)
self.assertEqual(0, r.returncode, r.stderr)
plan = json.load(open(out, encoding="utf-8"))
self.assertEqual("run-fixed-for-the-test", plan["runId"])
key = next(s["headers"]["Idempotency-Key"] for s in plan["requests"]
if s["op"] == "create")
self.assertTrue(key.endswith(ss._sha256_text("run-fixed-for-the-test")[:12]))
def test_a_harness_plan_still_has_no_publish_path(self): def test_a_harness_plan_still_has_no_publish_path(self):
steps = ss.plan_requests("docs/p/t/case/x.md", dict(SENT), None, None, steps = ss.plan_requests("docs/p/t/case/x.md", dict(SENT), None, None,
harness_test=True) harness_test=True)