fix(M02): 남의 층 쓰기 기록 = 성공한 쓰기만 — 길이 답한 뒤 남김 · 막힘 · 실패 부름 뺌(39-1d)

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_013rJwJMVL4EE31y77TB11AP

M02_MasterTemplete/M02_MasterTemplete_Router_Items.py

@@ -11,11 +11,14 @@ PLAN 27-1c) · `owner` = 회사 `{C}` · 개인 `{C}/{U}`(없으면 자기 층)
 from __future__ import annotations
 
 import logging
+from collections.abc import Callable, Coroutine
+from contextvars import ContextVar
 from pathlib import Path
 from typing import Any, Literal
 
 import anyio.from_thread
-from fastapi import APIRouter, Depends, HTTPException, Query
+from fastapi import APIRouter, Depends, HTTPException, Query, Request, Response
+from fastapi.routing import APIRoute
 from pydantic import BaseModel
 
 from common_util.common_util_audit import record_audit
@@ -37,7 +40,32 @@ from M02_MasterTemplete.M02_MasterTemplete_Router import (
 )
 
 logger = logging.getLogger(__name__)
-router = APIRouter(prefix="/api/m02", tags=["M02 MasterTemplete Items"])
+#: 이 요청이 쓴 남의 층(`others_write`) — 길이 성공해 답한 뒤에만 시스템 로그(`NotedRoute`)
+_WRITES: ContextVar[list[tuple[dict, access.Grant]] | None] = ContextVar("m02_writes", default=None)
+
+
+class NotedRoute(APIRoute):
+    """남의 층 쓰기 기록 = 성공한 쓰기만(39-1d) — 막히거나 실패한 부름(예외 · 4xx · 5xx)은 뺌."""
+
+    def get_route_handler(self) -> Callable[[Request], Coroutine[Any, Any, Response]]:
+        handler = super().get_route_handler()
+
+        async def run(request: Request) -> Response:
+            notes: list[tuple[dict, access.Grant]] = []
+            token = _WRITES.set(notes)
+            try:
+                response = await handler(request)
+            finally:
+                _WRITES.reset(token)
+            if response.status_code < 400:
+                for session, got in notes:
+                    await _note(session, got)
+            return response
+
+        return run
+
+
+router = APIRouter(prefix="/api/m02", tags=["M02 MasterTemplete Items"], route_class=NotedRoute)
 
 Layer = Literal["system", "company", "personal"]
 Session = dict[str, Any]
@@ -114,12 +142,17 @@ async def _note(session: Session, got: access.Grant) -> None:
 
 
 async def others_write(session: Session, got: access.Grant, folder: Path) -> None:
-    """남의 층 쓰기 — 처음(집계표 없음)이면 주인이 있는지 본 뒤 뼈대 · 쓸 때마다 시스템 로그."""
+    """남의 층 쓰기 — 처음(집계표 없음)이면 주인이 있는지 본 뒤 뼈대 · 시스템 로그는 길이 성공한
+    뒤(`NotedRoute` · 그 밖의 길은 바로)."""
     if not (folder / "table" / f"{store.STRUCTURE_TABLE}.json").is_file():
         if not await _member(str(got.owner)):
             raise HTTPException(status_code=403, detail="그 회사 · 그 회사 사람이 아님")
         items_store.ensure(folder)
-    await _note(session, got)
+    pending = _WRITES.get()
+    if pending is None:
+        await _note(session, got)
+    else:
+        pending.append((session, got))
 
 
 def _root(session: Session, layer: str, owner: str | None = None, write: bool = False) -> Path:

M02_MasterTemplete/M02_MasterTemplete_Router_Layers.py

@@ -34,7 +34,9 @@ from M02_MasterTemplete import M02_Template_Layers as layers
 
 logger = logging.getLogger(__name__)
 
-router = APIRouter(prefix="/api/m02", tags=["M02 MasterTemplete Layers"])
+router = APIRouter(
+    prefix="/api/m02", tags=["M02 MasterTemplete Layers"], route_class=items_router.NotedRoute
+)
 
 Layer = Literal["system", "company", "personal", "project"]
 

resources/tester/test_m02_layer_access.py

@@ -160,7 +160,7 @@ def test_남의_층_첫_쓰기는_소속_확인_뒤_뼈대(world: dict[str, Any]
     add["판"] = ""
     assert client.post("/api/m02/layers/company/tree?owner=9", json=add).status_code == 409
     assert (storage / "9/templates/table/구조물집계표.json").is_file()
-    assert [n[2] for n in world["notes"]] == ["7/44", "7/44", "9"]
+    assert [n[2] for n in world["notes"]] == ["7/44"]  # 성공한 쓰기만(39-1d) · 409 · 403 은 안 남김
 
 
 def test_남의_층_목록(world: dict[str, Any]):