ls tests/conftest.py tests/helpers/ tests/test_grc_*.py | head

# tests/test_task_survey_answer.py
import pytest

API = "/api/1.0/project-survey/answers"

@pytest.fixture
def seeded_task_survey(client, auth_headers):
    """建立一張 task_survey，回傳 uid 與題目清單。
    依 tests/helpers 內既有 seeding util；若無，直接呼 GRC 啟動專案的 fixture。
    """
    raise NotImplementedError("接 helper")  # 第一輪先讓 collection 就好


def test_put_twice_does_not_duplicate_rows(client, auth_headers, seeded_task_survey, db_session):
    ts_uid = seeded_task_survey["uid"]
    q_uid_1 = seeded_task_survey["question_uids"][0]

    payload_1 = {
        "status": 1,
        "answers": {q_uid_1: {"answer": "v1", "score": 0}},
    }
    r1 = client.put(f"{API}/{ts_uid}", json=payload_1, headers=auth_headers)
    assert r1.status_code == 200

    payload_2 = {
        "status": 1,
        "answers": {q_uid_1: {"answer": "v2", "score": 0}},  # 故意不帶 uid
    }
    r2 = client.put(f"{API}/{ts_uid}", json=payload_2, headers=auth_headers)
    assert r2.status_code == 200

    cnt = db_session.execute(
        "SELECT count(*) FROM survey.question_answers "
        "WHERE task_survey_id = :tid",
        {"tid": seeded_task_survey["id"]},
    ).scalar()
    assert cnt == 1, f"期望 1 row，實際 {cnt}（duplicate bug）"

pytest tests/test_task_survey_answer.py::test_put_twice_does_not_duplicate_rows -v

pytest tests/test_task_survey_answer.py::test_put_twice_does_not_duplicate_rows -v

git add tests/test_task_survey_answer.py
git commit -m "test(task-survey): add failing test for answer duplicate bug"

# 替換 question_answer_service.py:54-94 整段
questions = self.survey_question_domain_service.get_questions(survey_id=task_survey.survey_id)
q_uid_to_id = {q.uid: q.id for q in questions}

exist_answers = self.question_answer_domain_service.get_question_answers(
    QuestionAnswerQueryEntity(task_survey_id=task_survey.id)
)
existing_by_qid: dict[int, QuestionAnswerEntity] = {}
for a in exist_answers:
    cur = existing_by_qid.get(a.question_id)
    if cur is None or (a.updated_at and (cur.updated_at is None or a.updated_at > cur.updated_at)):
        existing_by_qid[a.question_id] = a

res_answer_list = []
for question_uid, answer in answers.items():
    question_id = q_uid_to_id.get(question_uid)
    if question_id is None:
        # 整張 survey 沒這題 — 不應該發生，跳過避免抽風
        continue

    payload_answer = answer.get("answer")
    payload_ext_answer = answer.get("ext_answer")
    payload_feedback = answer.get("feedback", "")
    payload_ext_feedback = answer.get("ext_feedback", "")
    payload_score = answer.get("score", 0)

    existing = existing_by_qid.get(question_id)
    if existing is not None:
        existing.score = payload_score
        existing.answer = payload_answer
        existing.ext_answer = payload_ext_answer
        existing.feedback = payload_feedback
        existing.ext_feedback = payload_ext_feedback
        existing.is_delete = 0
        existing.updated_user = user
        self.question_answer_domain_service.update_question_answer(existing)
        res_answer_list.append(existing)
    else:
        new_qa = QuestionAnswerEntity(
            score=payload_score,
            answer=payload_answer,
            ext_answer=payload_ext_answer,
            feedback=payload_feedback,
            ext_feedback=payload_ext_feedback,
            is_delete=0,
            question_id=question_id,
            survey_id=task_survey.survey_id,
            task_survey_id=task_survey.id,
            created_user=user,
            updated_user=user,
        )
        new_qa = self.question_answer_domain_service.add_question_answer(new_qa)
        existing_by_qid[question_id] = new_qa  # 防止同次 payload 同題重複造重複
        res_answer_list.append(new_qa)

pytest tests/test_task_survey_answer.py::test_put_twice_does_not_duplicate_rows -v

def test_put_twice_updates_existing_value(client, auth_headers, seeded_task_survey, db_session):
    ts_uid = seeded_task_survey["uid"]
    q_uid = seeded_task_survey["question_uids"][0]

    client.put(f"{API}/{ts_uid}", json={"status": 1, "answers": {q_uid: {"answer": "v1"}}}, headers=auth_headers)
    client.put(f"{API}/{ts_uid}", json={"status": 1, "answers": {q_uid: {"answer": "v2"}}}, headers=auth_headers)

    rows = db_session.execute(
        "SELECT answer FROM survey.question_answers "
        "WHERE task_survey_id = :tid AND question_id IN "
        "  (SELECT id FROM survey.survey_questions WHERE uid = :quid)",
        {"tid": seeded_task_survey["id"], "quid": q_uid},
    ).fetchall()
    assert len(rows) == 1
    assert rows[0][0] == "v2"

pytest tests/test_task_survey_answer.py -v

git add app/task_survey/service/question_answer_service.py tests/test_task_survey_answer.py
git commit -m "fix(task-survey): use natural key upsert in update_task_survey_answer

對應 docs/issues/pending/2026-04-28-task-survey-answer-duplicate.md 根因 §3.1。
不再用 row uid 做 new vs existing 判斷，改用 (task_survey_id, question_id)。"

def test_get_dedup_when_dirty_data_exists(client, auth_headers, seeded_task_survey, db_session):
    ts_id = seeded_task_survey["id"]
    q_id = seeded_task_survey["question_id_for"](seeded_task_survey["question_uids"][0])

    # 手插兩筆同 (task_survey_id, question_id)，模擬髒資料（migration 前狀態）
    # 注意：Phase 1.5 加 unique constraint 之後就插不進去；此測試在 1.5 完成後改用 raw SQL with ON CONFLICT DO NOTHING bypass，或拆成「constraint 還沒 enforce 前」的版本
    db_session.execute(
        "INSERT INTO survey.question_answers (task_survey_id, question_id, answer, created_user, updated_user, updated_at) "
        "VALUES (:tid, :qid, '\"old\"', 'seed', 'seed', NOW() - INTERVAL '1 hour')",
        {"tid": ts_id, "qid": q_id},
    )
    db_session.execute(
        "INSERT INTO survey.question_answers (task_survey_id, question_id, answer, created_user, updated_user, updated_at) "
        "VALUES (:tid, :qid, '\"new\"', 'seed', 'seed', NOW())",
        {"tid": ts_id, "qid": q_id},
    )
    db_session.commit()

    r = client.get(f"{API}/{seeded_task_survey['uid']}", headers=auth_headers)
    assert r.status_code == 200
    body = r.get_json()["data"]
    q_uid = seeded_task_survey["question_uids"][0]
    assert body[q_uid]["answer"] == "new"

pytest tests/test_task_survey_answer.py::test_get_dedup_when_dirty_data_exists -v

# infra/task_survey/repository/question_answer_repo_impl.py
from sqlalchemy import desc

class QuestionAnswerRepoImpl(...):
    ...
    def get_latest_by_task_survey_id(self, task_survey_id: int) -> list[QuestionAnswerEntity]:
        """以 (question_id) 為 partition、updated_at desc 取每組最新一筆。
        即使 DB 還有重複髒資料也能回對的版本。"""
        from sqlalchemy import select, func, text
        # PostgreSQL DISTINCT ON
        stmt = (
            select(self.model)
            .where(self.model.task_survey_id == task_survey_id)
            .order_by(self.model.question_id, desc(self.model.updated_at), desc(self.model.id))
            .distinct(self.model.question_id)
        )
        rows = self.session.execute(stmt).scalars().all()
        return self.mapper.to_list_entity(rows)

# domain/task_survey/repository/question_answer.py
@abstractmethod
def get_latest_by_task_survey_id(self, task_survey_id: int) -> list[QuestionAnswerEntity]: ...

# domain/task_survey/service/question_answer_domain_service.py
def get_latest_question_answers(self, task_survey_id: int) -> list[QuestionAnswerEntity]:
    return self.question_answer_repo.get_latest_by_task_survey_id(task_survey_id)

# app/task_survey/service/question_answer_service.py:40
res = self.question_answer_domain_service.get_latest_question_answers(task_survey.id)
res = QuestionAnswerResponseDTO.from_entity_list(res)
ans = {q.question_uid: q for q in res}
return ans

pytest tests/test_task_survey_answer.py -v

git add infra/ domain/ app/task_survey/service/question_answer_service.py tests/
git commit -m "fix(task-survey): use DISTINCT ON for GET to ignore duplicate rows"

-- Date: 2026-04-28
-- Task Survey 答案表去重 + 加 UNIQUE constraint
-- 對應 issue: docs/issues/pending/2026-04-28-task-survey-answer-duplicate.md
-- 對應 plan : docs/features/FR-021-2604-survey-answer-duplicate-fix/implementation-plan.md (Phase 1.4)

-- 1. 備份重複資料以利稽核（保留 7 天再 drop） (2026-04-28)
CREATE TABLE IF NOT EXISTS survey.question_answers_dedup_backup_20260428 AS
SELECT * FROM survey.question_answers WHERE 1=0;

-- 2. 把要刪除的（同 task_survey_id+question_id 內非最新）寫進備份 (2026-04-28)
WITH ranked AS (
    SELECT id,
           ROW_NUMBER() OVER (
               PARTITION BY task_survey_id, question_id
               ORDER BY updated_at DESC NULLS LAST, id DESC
           ) AS rn
    FROM survey.question_answers
)
INSERT INTO survey.question_answers_dedup_backup_20260428
SELECT qa.* FROM survey.question_answers qa
JOIN ranked r ON r.id = qa.id
WHERE r.rn > 1;

-- 3. 刪除非最新的重複 row (2026-04-28)
WITH ranked AS (
    SELECT id,
           ROW_NUMBER() OVER (
               PARTITION BY task_survey_id, question_id
               ORDER BY updated_at DESC NULLS LAST, id DESC
           ) AS rn
    FROM survey.question_answers
)
DELETE FROM survey.question_answers qa
USING ranked r
WHERE qa.id = r.id AND r.rn > 1;

-- 4. 加 UNIQUE constraint (2026-04-28)
ALTER TABLE survey.question_answers
    ADD CONSTRAINT uq_question_answers_task_survey_question
    UNIQUE (task_survey_id, question_id);

-- 5. 驗證：每組 (task_survey_id, question_id) 必須剛好 1 筆 (2026-04-28)
DO $$
DECLARE
    dup_count integer;
BEGIN
    SELECT COUNT(*) INTO dup_count FROM (
        SELECT 1 FROM survey.question_answers
        GROUP BY task_survey_id, question_id
        HAVING COUNT(*) > 1
    ) g;
    IF dup_count > 0 THEN
        RAISE EXCEPTION 'Dedup failed: % duplicate groups remain', dup_count;
    END IF;
END $$;

PGPASSWORD='jedi@123!' psql -h 192.168.50.188 -p 25432 -U cmmgr -d guidant_ai_stg \
    -f scripts/sql/2026-04-28-survey-answer-dedup.sql

PGPASSWORD='jedi@123!' psql -h 192.168.50.188 -p 25432 -U cmmgr -d guidant_ai_stg -c "
SELECT
  (SELECT COUNT(*) FROM survey.question_answers) AS total_rows,
  (SELECT COUNT(*) FROM (SELECT 1 FROM survey.question_answers GROUP BY task_survey_id, question_id HAVING COUNT(*) > 1) g) AS duplicate_pairs,
  (SELECT COUNT(*) FROM survey.question_answers_dedup_backup_20260428) AS backup_rows;
"

# 改用 ON CONFLICT DO NOTHING + 暫時 disable trigger 是過度。
# 最簡單：把這個 case 改用 raw INSERT bypass constraint 是不可能的。
# 改測試方向：「即使透過 service 多次寫入，GET 也只回最新版本」，這已經被 test_put_twice_updates_existing_value 涵蓋。
# 結論：刪除 test_get_dedup_when_dirty_data_exists，改保留 test_put_twice_updates_existing_value 即可。

pytest tests/test_task_survey_answer.py -v

git add scripts/sql/2026-04-28-survey-answer-dedup.sql tests/test_task_survey_answer.py
git commit -m "fix(task-survey): dedup migration + UNIQUE(task_survey_id, question_id)

dev DB 跑後 2338→602 rows，1736 重複 row 進備份表 question_answers_dedup_backup_20260428。"

def test_socket_full_data_path_does_not_duplicate(client, auth_headers, seeded_task_survey, db_session, container):
    """socket on_update 內部呼叫 update_task_survey_answer(room, None, full_data, user)，
    多次呼叫不應產生重複 row（Phase 1 service 修完後）。"""
    qa_service = container.question_answer_container.question_answer_service()
    ts_uid = seeded_task_survey["uid"]
    q_uid = seeded_task_survey["question_uids"][0]

    full_data_1 = {q_uid: {"answer": "via socket v1"}}
    qa_service.update_task_survey_answer(ts_uid, None, full_data_1, "test-user")

    full_data_2 = {q_uid: {"answer": "via socket v2"}}
    qa_service.update_task_survey_answer(ts_uid, None, full_data_2, "test-user")

    cnt = db_session.execute(
        "SELECT count(*) FROM survey.question_answers "
        "WHERE task_survey_id = :tid",
        {"tid": seeded_task_survey["id"]},
    ).scalar()
    assert cnt == 1

pytest tests/test_task_survey_answer.py -v

git add tests/test_task_survey_answer.py
git commit -m "test(task-survey): cover socket on_update path against duplicate"

---
type: fix
modules: [task_survey]
issue: docs/issues/pending/2026-04-28-task-survey-answer-duplicate.md
commit: <填 phase1 最後一筆 commit hash>
---

# fix(task-survey): 答案重複寫入 bug Phase 1（natural key upsert + DB 去重）

## 為什麼

`POST /project-survey/answers/<uid>` PUT 與 socket on_update 共用的 `update_task_survey_answer` 用 row uid 判斷新舊，payload 沒帶 uid（或對不上）就 INSERT，造成同 (task_survey_id, question_id) 多筆 row。dev DB 累積 1736 筆重複（71%）。對應 Billows NICS 已修過的 pattern A，跨專案根因相同。

## 變更

- `app/task_survey/service/question_answer_service.py`
  - `update_task_survey_answer` 改用 `(task_survey_id, question_id)` natural key upsert
  - `get_answer_list_by_task_survey_uid` 改用新 repo method `get_latest_question_answers` 走 SQL DISTINCT ON 去重
- `infra/task_survey/repository/question_answer_repo_impl.py` 新增 `get_latest_by_task_survey_id`
- `domain/task_survey/repository/question_answer.py` 新增抽象方法
- `domain/task_survey/service/question_answer_domain_service.py` wrap
- `scripts/sql/2026-04-28-survey-answer-dedup.sql`
  - 把同組非最新 row 寫進備份表 `survey.question_answers_dedup_backup_20260428`
  - 刪除這些 row
  - 加 `UNIQUE (task_survey_id, question_id)` constraint
  - DO block 驗證跑完後沒有重複組

## API 變更

無外部合約變動。GET 回傳的 dict 結構不變、PUT 接收的 payload 結構不變。

## 測試結果

git add docs/changelog/2026-04-28-fix-task-survey-answer-duplicate-phase1.md
git commit -m "docs(changelog): task-survey answer duplicate fix phase 1"

def test_patch_single_question_upserts_one_row(client, auth_headers, seeded_task_survey, db_session, container):
    qa_service = container.question_answer_container.question_answer_service()
    ts_uid = seeded_task_survey["uid"]
    q_uid = seeded_task_survey["question_uids"][0]

    qa_service.patch_task_survey_answer(ts_uid, q_uid, {"answer": "patched v1"}, "test-user")
    qa_service.patch_task_survey_answer(ts_uid, q_uid, {"answer": "patched v2"}, "test-user")

    rows = db_session.execute(
        "SELECT answer FROM survey.question_answers "
        "WHERE task_survey_id = :tid",
        {"tid": seeded_task_survey["id"]},
    ).fetchall()
    assert len(rows) == 1
    assert rows[0][0] == "patched v2"

pytest tests/test_task_survey_answer.py::test_patch_single_question_upserts_one_row -v

# app/task_survey/service/question_answer_service.py
@transaction
def patch_task_survey_answer(
    self,
    task_survey_uid: str,
    question_uid: str,
    answer_patch: dict,
    user: str,
) -> QuestionAnswerEntity:
    """單題 upsert。供 socket on_update 與 REST patch endpoint 使用。

    與 update_task_survey_answer 不同：
    - 不寫 history（history 只在 PUT 暫存/提交時寫）
    - 不變更 task_survey.status（status 由暫存/提交按鈕掌控）
    - 只動一題，N+1 也不存在
    """
    task_survey = self.task_survey_domain_service.get_task_survey_by_uid(task_survey_uid)
    if not task_survey:
        raise NotFound(ErrorCode.TASK_SURVEY_NOT_FOUND)

    question = next(
        (q for q in self.survey_question_domain_service.get_questions(survey_id=task_survey.survey_id)
         if q.uid == question_uid),
        None,
    )
    if question is None:
        raise NotFound(ErrorCode.TASK_SURVEY_QUESTION_NOT_FOUND)

    existing = self.question_answer_domain_service.get_question_answer(
        QuestionAnswerQueryEntity(task_survey_id=task_survey.id, question_id=question.id)
    )

    if existing is not None:
        existing.answer = answer_patch.get("answer", existing.answer)
        existing.ext_answer = answer_patch.get("ext_answer", existing.ext_answer)
        existing.feedback = answer_patch.get("feedback", existing.feedback)
        existing.ext_feedback = answer_patch.get("ext_feedback", existing.ext_feedback)
        existing.score = answer_patch.get("score", existing.score)
        existing.is_delete = 0
        existing.updated_user = user
        return self.question_answer_domain_service.update_question_answer(existing)

    new_qa = QuestionAnswerEntity(
        score=answer_patch.get("score", 0),
        answer=answer_patch.get("answer"),
        ext_answer=answer_patch.get("ext_answer"),
        feedback=answer_patch.get("feedback", ""),
        ext_feedback=answer_patch.get("ext_feedback", ""),
        is_delete=0,
        question_id=question.id,
        survey_id=task_survey.survey_id,
        task_survey_id=task_survey.id,
        created_user=user,
        updated_user=user,
    )
    return self.question_answer_domain_service.add_question_answer(new_qa)

# common/code/error_code.py
TASK_SURVEY_QUESTION_NOT_FOUND = ("問卷題目不存在", "TASK_SURVEY_404006")

pytest tests/test_task_survey_answer.py::test_patch_single_question_upserts_one_row -v

git add app/task_survey/service/question_answer_service.py common/code/error_code.py tests/test_task_survey_answer.py
git commit -m "feat(task-survey): add patch_task_survey_answer for single-question upsert"

def test_post_answers_patch_endpoint(client, auth_headers, seeded_task_survey, db_session):
    ts_uid = seeded_task_survey["uid"]
    q_uid = seeded_task_survey["question_uids"][0]

    payload = {"patches": [{"question_uid": q_uid, "answer": {"answer": "rest-patch"}}]}
    r = client.post(f"{API}/{ts_uid}/patch", json=payload, headers=auth_headers)
    assert r.status_code == 200

    cnt = db_session.execute(
        "SELECT count(*) FROM survey.question_answers "
        "WHERE task_survey_id = :tid",
        {"tid": seeded_task_survey["id"]},
    ).scalar()
    assert cnt == 1

# api/task_survey/serializers/question_answer_patch.py
from marshmallow import Schema, fields

class _PatchAnswerInner(Schema):
    answer = fields.Raw(required=False, allow_none=True)
    ext_answer = fields.Raw(required=False, allow_none=True)
    feedback = fields.Raw(required=False, allow_none=True)
    ext_feedback = fields.Raw(required=False, allow_none=True)
    score = fields.Integer(required=False)


class PatchItemSchema(Schema):
    question_uid = fields.String(required=True)
    answer = fields.Nested(_PatchAnswerInner, required=True)


class PatchRequestSchema(Schema):
    patches = fields.List(fields.Nested(PatchItemSchema), required=True)

# api/task_survey/routes/question_answer_route.py 新增
class TaskSurveyAnswerPatchRoute(MethodResource):
    @doc(description='批次單題 patch（供 socket / keepalive flush 用）',
         tags=['Main Project Survey Answer'], params=AUTH_PARAMS)
    @jwt_required()
    def post(self, uid,
             question_answer_service: QuestionAnswerService = Provide[
                 Containers.question_answer_container.question_answer_service]):
        user = get_user_context().login_name
        payload = request.get_json(silent=True) or {}
        patches = payload.get("patches", [])
        for patch in patches:
            question_answer_service.patch_task_survey_answer(
                uid, patch["question_uid"], patch.get("answer", {}), user
            )
        return return_response(True, {"applied": len(patches)})

# api/task_survey/__init__.py 在 TaskSurveysAnswersRoute 之後
from api.task_survey.routes.question_answer_route import TaskSurveyAnswerPatchRoute
api.add_resource(TaskSurveyAnswerPatchRoute, '/project-survey/answers/<string:uid>/patch')

pytest tests/test_task_survey_answer.py::test_post_answers_patch_endpoint -v

git add api/task_survey/
git commit -m "feat(task-survey): add POST /project-survey/answers/<uid>/patch"

def on_update(self, data):
    """問卷填答案更新（單題 patch）。

    舊版接收 full_data 整份覆蓋 + Redis snapshot，會造成 stale data 與重複寫入。
    新版只動被改的那題，避免多人協作覆蓋彼此編輯。
    """
    user = data.get('user')
    room = data.get('room')
    receive_data = data.get('data') or {}
    question_uid = receive_data.get('question_uid')
    answer_patch = receive_data.get('answer') or {}

    if not question_uid:
        emit('error', {'msg': 'question_uid is required'})
        return

    # 寫 DB（單題 upsert）
    self.question_answer_service.patch_task_survey_answer(room, question_uid, answer_patch, user)

    # broadcast（payload 維持原 shape，前端不需動）
    result = {
        'msg': f' {user} 更新資料',
        'user': user,
        'room': room,
        'uid': receive_data.get('uid'),
        'data': receive_data,
    }
    emit('updated', result, room=room)

def on_join(self, data):
    user = data.get('user')
    room = data.get('room')
    data["event"] = "join"

    redis_cli = RedisClient()
    redis_cli.ping()
    key = f'fill-survey:{room}:users'
    users = redis_cli.get_all_from_list(key)
    if user not in users:
        redis_cli.lpush(key, user)
    users = redis_cli.get_all_from_list(key)

    result = {
        'msg': f'{user} 加入編輯',
        'users': users,
        'user': user,
        'data': None,
    }

    # 多人時，從 DB 拉最新 answer dict（取代舊 Redis snapshot）
    if len(users) > 1:
        result['data'] = self.question_answer_service.get_answer_list_by_task_survey_uid(room)

    join_room(room)
    emit('joined', result, room=room)

def on_leave(self, data):
    user = data.get('user')
    room = data.get('room')
    data["event"] = "leave"

    redis_cli = RedisClient()
    key = f'fill-survey:{room}:users'
    users = redis_cli.get_all_from_list(key)
    if user in users:
        redis_cli.lrem(key, user)
    users = redis_cli.get_all_from_list(key)

    # 不再需要清 fill_survey:<room>:updates（已不寫入）

    leave_room(room)
    emit('left', {'msg': f'{user} 離開編輯', 'users': users, 'user': user}, room=room)

pytest tests/test_task_survey_answer.py -v

git add app/task_survey/handler/fill_survey_socketio_handler.py
git commit -m "fix(task-survey): socket on_update uses single-question patch, on_join reads from DB

移除 fill_survey:<room>:updates Redis stale snapshot 模式。
對應 Billows R1 / R2 修法。"

grep -rn "fill_survey:.*:updates" app/ infra/ domain/

redis-cli -h 192.168.50.189 KEYS "fill_survey:*:updates" | xargs -I{} redis-cli -h 192.168.50.189 DEL {}

# domain/task_survey/service/question_answer_domain_service.py
def update_question_answer_in_memory(self, entity: QuestionAnswerEntity) -> QuestionAnswerEntity:
    """已 in-memory 確認 entity 來自 DB，不再跑 verify_*。"""
    return self.question_answer_repo.update(entity)

def test_put_100_questions_under_3s(client, auth_headers, seeded_100q_task_survey):
    import time
    ts_uid = seeded_100q_task_survey["uid"]
    answers = {q_uid: {"answer": f"v_{i}"} for i, q_uid in enumerate(seeded_100q_task_survey["question_uids"])}

    t0 = time.time()
    r = client.put(f"{API}/{ts_uid}", json={"status": 1, "answers": answers}, headers=auth_headers)
    elapsed = time.time() - t0

    assert r.status_code == 200
    assert elapsed < 3.0, f"100 題 PUT 跑 {elapsed:.2f}s，超過 3s 上限"

@transaction
def revert_question_answer_from_history(self, task_survey_uid, history_uid, user) -> bool:
    task_survey = self.task_survey_domain_service.get_task_survey_by_uid(task_survey_uid)
    answer_history = self.question_answer_history_domain_service.get_question_answer_history(history_uid)
    history_answers = self.question_answer_history_detail_domain_service.get_history_details(
        QuestionAnswerHistoryDetailQueryEntity(history_id=answer_history.id)
    )

    # 預取一次當前 answers，建 map
    current_answers = self.question_answer_domain_service.get_question_answers(
        QuestionAnswerQueryEntity(task_survey_id=task_survey.id)
    )
    current_by_qid = {a.question_id: a for a in current_answers}

    history_detail_entity_list = []
    for answer in history_answers:
        question_answer = current_by_qid.get(answer.question_id)
        if question_answer is None:
            continue  # 該題已被刪 — skip

        question_answer.score = answer.score
        question_answer.answer = answer.answer
        question_answer.ext_answer = answer.ext_answer
        question_answer.feedback = answer.feedback
        question_answer.ext_feedback = answer.ext_feedback
        question_answer.updated_user = user
        self.question_answer_domain_service.update_question_answer_in_memory(question_answer)

        history_detail_entity_list.append(
            QuestionAnswerHistoryDetailEntity(
                task_survey_id=task_survey.id,
                question_id=answer.question_id,
                score=answer.score,
                answer=answer.answer,
                ext_answer=answer.ext_answer,
                feedback=answer.feedback,
                ext_feedback=answer.ext_feedback,
                updated_user=user,
                created_user=user,
            )
        )

    history_entity = QuestionAnswerHistoryEntity(
        task_survey_id=task_survey.id,
        type="R",
        updated_user=user,
        created_user=user,
    )
    self.question_answer_history_domain_service.create_history_with_details(
        history_entity, history_detail_entity_list
    )
    return True

def test_revert_100_questions_under_3s(...):  # 同上 pattern

pytest tests/test_task_survey_answer.py -v
pytest tests/  # 確認沒打到別處

PGPASSWORD='jedi@123!' psql -h 192.168.50.188 -p 25432 -U cmmgr -d guidant_ai_stg -c "
SELECT COUNT(*) FROM (
    SELECT 1 FROM survey.question_answers
    GROUP BY task_survey_id, question_id HAVING COUNT(*) > 1
) g;
"

## ⓪ Resolution

| 項目 | 內容 |
|------|------|
| 修復日期 | 2026-04-28 |
| Commits | <Phase1>, <Phase2>, <Phase3> short hashes |
| Changelogs | docs/changelog/2026-04-28-fix-task-survey-answer-duplicate-phase1.md / phase2 / phase3 |
| Follow-up | 前端是否帶 row uid 不是 root cause（後端已防禦），但建議讓 fe 補上 dirty tracking + keepalive flush 對應 Billows fe 修法 |

git mv docs/issues/pending/2026-04-28-task-survey-answer-duplicate.md \
       docs/issues/resolved/2026-04-28-task-survey-answer-duplicate.md

git commit -m "docs(issue): mark task-survey answer duplicate fix resolved"