# Task Survey Answer Checkpoint Optimization — Implementation Plan

> **For agentic workers:** REQUIRED SUB-SKILL: Use `superpowers:subagent-driven-development` to execute task-by-task.

**Goal:** 把暫存/提交的整份 `answers` dict PUT 改成單題 patches 的 checkpoint 模式（Billows 已 production 驗證的設計），順便修掉 reactivity bug 與 socket broadcast shape 不一致的問題。100+ 題問卷的儲存從「每次寫整份」變「只寫 dirty 的題」，避免 DB 寫入塞車。

**Architecture:**
- BE：新增 `POST /api/1.0/project-survey/answers/<uid>/checkpoint` endpoint。內部複用 Phase 1-3 的自然鍵 upsert + DISTINCT ON + in-memory update 等 primitives；新增「fill empty rows for missing questions」邏輯（暫存/提交時保留原本「補空 row」語意）；單一 transaction 寫 history。改 socket on_update broadcast shape 為純 answer body（與 Billows 一致）。舊 PUT `/answers/<uid>` 保留但 deprecate（Excel 批次匯入仍用）。
- FE：`setAnswer` 改全物件重新指派（修 reactivity bug #1）。`dirtyAnswers` 改成「所有未 checkpoint 的改動」。暫存/提交按鈕改打 `/checkpoint` + dirty patches。`socket.on('updated')` 對齊新 shape。`socket.on('joined')` 加 dirty-aware merge。
- jedi-*：本次 plan 不動。如效能仍不足，後續另外進版 `jedi-common.BaseRepositoryImpl.update_attached` 與 `jedi-survey.get_questions selectinload`。

**Tech Stack:** Flask + Flask-RESTful + Flask-SocketIO + SQLAlchemy + jedi-common (`@transaction` / `BaseRepositoryImpl`) + jedi-survey + Vue 3 + PrimeVue + socket.io-client + Pinia + axios.

**Reference:**
- 對應 issue：`docs/issues/resolved/2026-04-28-task-survey-answer-duplicate.md`（會在 Phase 7 收尾追加 follow-up 段）
- 對應 BE Phase 1-3 plan：`docs/features/FR-021-2604-survey-answer-duplicate-fix/implementation-plan.md`
- Billows BE checkpoint：`/Users/chouraymond/Projects/Billows/NICS/billows-be/app/survey/service/main_project_survey_service.py:108-271 checkpoint_main_project_survey_answer`
- Billows BE socket broadcast shape：`/Users/chouraymond/Projects/Billows/NICS/billows-be/app/survey/handler/fill_survey_socketio_handler.py:109-144 on_update`
- Billows FE setter：`/Users/chouraymond/Projects/Billows/NICS/billows-fe/src/views/survey/survey-manage/EvaluationView.vue:455-488 updateAnswer + onAnswerInput`
- Billows FE socket receivers：同檔 line 156-185
- Billows FE checkpoint usage：同檔 line 560-580 onTempSaveClick + 552-558 buildPatchesFromDirty

---

## File Structure

### BE (`compliance-manager-be`)

| 檔案 | 動作 | 責任 |
|------|------|------|
| `app/task_survey/service/question_answer_service.py` | 改 | 新增 `checkpoint_task_survey_answer(uid, status, patches, user)`；保留現有 `update_task_survey_answer`（PUT 路徑沿用） |
| `api/task_survey/serializers/question_answer_checkpoint.py` | 新增 | `_PatchAnswerInner`/`PatchItemSchema`/`CheckpointRequestSchema` (`{status, patches}`) |
| `api/task_survey/routes/question_answer_route.py` | 改 | 新增 `TaskSurveyAnswerCheckpointRoute` |
| `api/task_survey/__init__.py` | 改 | 註冊 `POST /project-survey/answers/<string:uid>/checkpoint` |
| `app/task_survey/handler/fill_survey_socketio_handler.py` | 改 | `on_update` broadcast `result['data'] = answer_patch`（純答案 body）+ `result['question_uid']`（top-level 給 FE 直接用），對齊 Billows shape |
| `tests/test_task_survey_answer.py` | 改 | 加 `checkpoint` 系列 tests |
| `docs/changelog/2026-04-30-tweak-task-survey-checkpoint-endpoint.md` | 新增 | Phase 4 BE changelog |

### FE (`compliance-manager-fe`)

| 檔案 | 動作 | 責任 |
|------|------|------|
| `src/config/api/api.js` | 改 | 加 `PROJECT_SURVEY_ANSWERS_CHECKPOINT` 常數（與 PATCH 共用 base，後綴 `/checkpoint`） |
| `src/views/survey-v2/composables/useSurveySocket.js` | 改 | `socket.on('updated')` 對齊新 broadcast shape；加 dirty-awareness（caller 提供 dirty 判斷 callback）；`socket.on('joined')` 改 dirty-aware merge |
| `src/views/survey-v2/SurveyPreview.vue` | 改 | `setAnswer` 全物件重新指派；`dirtyAnswers` 永久保留所有改動到 checkpoint 成功為止；`onTempSave`/`onSave` 改打 `/checkpoint` + `buildPatchesFromDirty()`；移除大部分 `clearDirty` 呼叫；socket-side dirty 判斷傳給 useSurveySocket |
| `docs/changelog/survey-fill-collab-fix/20260430_*_implementation-plan.md` | 新增 | FE plan |
| `docs/changelog/survey-fill-collab-fix/20260430_*_checkpoint-changelog.md` | 新增 | Phase 5 FE changelog |

---

## Conventions（每個 task 都套用）

- BE：服務層 public method 一律 `@transaction`；不在 helper 內 `get_session()`；DDD 嚴格、Route 不查 DB。
- BE serializer / route 加 `@use_kwargs(...)` 做 marshmallow 驗證，缺欄位回 422。
- BE migration 不需要（無 schema 變動）。
- BE error code 沿用既有 `TASK_SURVEY_*`，必要時新增 `TASK_SURVEY_400xxx`。
- FE：4-space indent；Composition API + `<script setup>`；i18n key 走 `lang.*`；Loading 狀態必備（暫存按鈕已有 `submitted` ref）；BaseService 走 axios。
- FE：dirty Map 內存清楚的 cloned snapshot（避免 plain-object reference 跟 Vue reactive proxy 混淆）。
- 所有 commit 不自動跑（user explicitly approve commits per task）。
- 每個 phase 結束有 changelog（CLAUDE.md 規定）。

---

# Phase 4 — BE: 加 `/checkpoint` endpoint + 對齊 broadcast shape

**Outcome:**
- `POST /api/1.0/project-survey/answers/<uid>/checkpoint` 上線、可接 `{status, patches}`
- 100 題 checkpoint 在單一 transaction 內完成、寫一次 history、補齊空 row、更新 status
- socket on_update broadcast `data` 是純 answer body（不再是 wrapping dict）
- 整套 service-level pytest 全綠、5 次連跑無 flake
- BE changelog 寫好

## Task 4.1：新 service `checkpoint_task_survey_answer`

**Files:**
- Modify: `app/task_survey/service/question_answer_service.py`

### Spec

```python
@transaction
def checkpoint_task_survey_answer(
    self,
    task_survey_uid: str,
    status: int,
    patches: list,        # [{question_uid: str, answer: dict}, ...]
    user: str,
) -> TaskSurveyDTO:
    """暫存/提交 checkpoint，單一 transaction 完成所有寫入。

    流程：
    1. 取 task_survey、survey questions、existing answers map（自然鍵）
    2. 套 patches（in-memory mutation 已存在的 entity，或 ADD 新的）
    3. 補空 row（survey 有題目但 DB 沒 row）— 保留舊 PUT 路徑的「全題目都有 row」語意，便於 revert
    4. 更新 task_survey.status（不可從 status=2/9 回退到較早狀態）
    5. 寫 history + history_details（從 in-memory state，不重 SELECT）
    6. 通知 status change（如有變動）

    與 update_task_survey_answer 差別：
    - 接收 patches list（單題 dict）而非 answers dict（整份覆蓋語意）
    - 走 update_question_answer_in_memory 跳 verify SELECT
    - history details 從 in-memory state，省一輪 SELECT
    """
    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)

    questions = self.survey_question_domain_service.get_questions(survey_id=task_survey.survey_id)
    q_uid_to_obj = {q.uid: q for q in questions}

    exist_answers = self.question_answer_domain_service.get_question_answers(
        QuestionAnswerQueryEntity(task_survey_id=task_survey.id)
    )
    # 重複時保留 updated_at 最新（migration 後不該再有，但服務層仍防禦）
    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 = []

    # 1. 套 patches
    for patch in (patches or []):
        question_uid = patch.get("question_uid")
        answer_patch = patch.get("answer") or {}
        if not question_uid:
            continue
        question = q_uid_to_obj.get(question_uid)
        if question is None:
            continue  # survey 沒這題，跳過

        existing = existing_by_qid.get(question.id)
        if existing is not None:
            if "answer" in answer_patch:
                existing.answer = answer_patch["answer"]
            if "ext_answer" in answer_patch:
                existing.ext_answer = answer_patch["ext_answer"]
            if "feedback" in answer_patch:
                existing.feedback = answer_patch["feedback"]
            if "ext_feedback" in answer_patch:
                existing.ext_feedback = answer_patch["ext_feedback"]
            if "score" in answer_patch:
                existing.score = answer_patch["score"]
            existing.updated_user = user
            self.question_answer_domain_service.update_question_answer_in_memory(existing)
            res_answer_list.append(existing)
        else:
            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", ""),
                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
            res_answer_list.append(new_qa)

    # 2. 補空 row（survey 有題目但 DB 沒 row 且這次沒被 patch）
    new_empty_qas = []
    for question in questions:
        if question.id in existing_by_qid:
            continue
        empty_qa = QuestionAnswerEntity(
            score=0,
            answer=None,
            ext_answer=None,
            feedback="",
            ext_feedback="",
            question_id=question.id,
            survey_id=task_survey.survey_id,
            task_survey_id=task_survey.id,
            created_user=user,
            updated_user=user,
        )
        empty_qa = self.question_answer_domain_service.add_question_answer(empty_qa)
        existing_by_qid[question.id] = empty_qa
        new_empty_qas.append(empty_qa)
        res_answer_list.append(empty_qa)

    # 3. 更新 status（含通知）
    survey_status_change = False
    if status is not None:
        survey_status_change = (status != task_survey.status)
        task_survey.status = status

    task_survey_entity = self.task_survey_domain_service.update_task_survey(task_survey)

    if survey_status_change:
        self.task_survey_service.notify_task_survey_assignee_status_change(task_survey.uid)

    # 4. 寫 history
    history = QuestionAnswerHistoryEntity(
        task_survey_id=task_survey.id,
        updated_user=user,
        created_user=user,
    )
    history_details = []
    for answer in res_answer_list:
        history_details.append(
            QuestionAnswerHistoryDetailEntity(
                task_survey_id=task_survey.id,
                survey_id=task_survey.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,
            )
        )
    self.question_answer_history_domain_service.create_history_with_details(history, history_details)

    return TaskSurveyDTO.from_entity(task_survey_entity)
```

**注意：** 不要動既有 `update_task_survey_answer`（PUT 路徑沿用、Excel 批次匯入仍走它）。

### Steps

- [ ] **Step 1**: Read `app/task_survey/service/question_answer_service.py` 整檔，理解 imports 與既有 `update_task_survey_answer` 的 history 寫入區塊。
- [ ] **Step 2**: 加 `checkpoint_task_survey_answer` method（依 spec），放在 `update_task_survey_answer` 與 `survey_has_answer` 之間。
- [ ] **Step 3**: Confirm DDD discipline — 沒 `from infra.*` import、沒 `get_session()`、沒 `self.session`。
- [ ] **Step 4**: Commit。Message:
  ```
  feat(task-survey): add checkpoint_task_survey_answer for patches-based save

  暫存/提交改成接收 dirty patches list 而非整份 answers dict。
  - 套 patches → 補空 row（survey 有題目但 DB 沒 row 且這次沒 patch 到）→ 更新 status
  - 重用 update_question_answer_in_memory 消 N+1 SELECT
  - 寫 history 從 in-memory state，不重 SELECT
  - 單一 @transaction，整段流程一次 commit

  舊 update_task_survey_answer 保留（PUT 路徑沿用、Excel 批次匯入仍走它）。

  對應 plan: docs/features/FR-023-2604-survey-answer-checkpoint-optimization/implementation-plan.md (Phase 4 Task 4.1)
  ```

---

## Task 4.2：新 serializer `CheckpointRequestSchema`

**Files:**
- Create: `api/task_survey/serializers/question_answer_checkpoint.py`

### Spec

```python
"""Request schema for POST /project-survey/answers/<uid>/checkpoint."""
from marshmallow import Schema, fields


class _PatchAnswerInner(Schema):
    """Inner answer body — all fields optional (PATCH semantics)."""
    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 CheckpointRequestSchema(Schema):
    status = fields.Integer(required=True)
    patches = fields.List(fields.Nested(PatchItemSchema), required=True)
```

### Steps

- [ ] **Step 1**: Create file with content above.
- [ ] **Step 2**: Commit:
  ```
  feat(task-survey): add CheckpointRequestSchema serializer

  對應 plan: ... (Phase 4 Task 4.2)
  ```

---

## Task 4.3：新 route `TaskSurveyAnswerCheckpointRoute` + 註冊

**Files:**
- Modify: `api/task_survey/routes/question_answer_route.py`
- Modify: `api/task_survey/__init__.py`

### Spec — route handler

```python
class TaskSurveyAnswerCheckpointRoute(MethodResource):
    @doc(description='暫存/提交問卷答案（patches-based）',
         tags=['Main Project Survey Answer'], params=AUTH_PARAMS)
    @use_kwargs(CheckpointRequestSchema, location='json', apply=True)
    @jwt_required()
    def post(self, uid, status, patches,
             question_answer_service: QuestionAnswerService = Provide[
                 Containers.question_answer_container.question_answer_service]):
        user = get_user_context().login_name
        data = question_answer_service.checkpoint_task_survey_answer(
            uid, status, patches, user
        )
        return return_response(True, data)
```

### Spec — 註冊

```python
# api/task_survey/__init__.py 內 create_module():
from api.task_survey.routes.question_answer_route import (
    TaskSurveysAnswersRoute,
    TaskSurveyAnswerPatchRoute,
    TaskSurveyAnswerCheckpointRoute,
)
...
api.add_resource(
    TaskSurveyAnswerCheckpointRoute,
    '/project-survey/answers/<string:uid>/checkpoint'
)
```

### Steps

- [ ] **Step 1**: 看現有 `TaskSurveyAnswerPatchRoute` 的 imports，確認 `use_kwargs` 已 import；加 `from api.task_survey.serializers.question_answer_checkpoint import CheckpointRequestSchema`。
- [ ] **Step 2**: 加 route 類別。
- [ ] **Step 3**: 改 `__init__.py` 的 import + register。
- [ ] **Step 4**: 跑 url_map 確認註冊：
  ```bash
  .venv/bin/python -c "
  import os, sys
  sys.path.insert(0, '.')
  sys.setrecursionlimit(5000)
  os.environ.setdefault('GITHUB_PRIVATE_TOKEN', 'fake')
  os.environ.setdefault('GITLAB_PRIVATE_TOKEN', 'fake')
  os.environ.setdefault('GITLAB_URL', 'https://gitlab.example.com')
  os.environ.setdefault('GITLAB_API_VERSION', '4')
  from dotenv import load_dotenv
  load_dotenv('.env.test', override=True)
  from core.app_factory import create_app
  from api.task_survey import create_module
  app = create_app(enable_socketio=False)
  app.register_blueprint(create_module())
  rules = sorted([r.rule for r in app.url_map.iter_rules() if 'project-survey/answers' in r.rule])
  for r in rules:
      print(r)
  "
  ```
  預期輸出含：
  ```
  /api/1.0/project-survey/answers/<string:uid>
  /api/1.0/project-survey/answers/<string:uid>/checkpoint
  /api/1.0/project-survey/answers/<string:uid>/patch
  ```
- [ ] **Step 5**: Commit：
  ```
  feat(task-survey): add POST /project-survey/answers/<uid>/checkpoint

  暫存/提交專用 endpoint，接收 {status, patches}。FE 改打這條後可避免
  PUT 整份 answers dict 的 DB 寫入塞車（100+ 題每次儲存只送 dirty）。

  對應 plan: ... (Phase 4 Task 4.3)
  ```

---

## Task 4.4：BE socket on_update broadcast shape 改

**Files:**
- Modify: `app/task_survey/handler/fill_survey_socketio_handler.py`

### Spec

舊：
```python
result = {
    'msg': f' {user} 更新資料',
    'user': user,
    'room': room,
    'uid': receive_data.get('uid'),
    'data': receive_data,  # 整個 receive_data 含 question_uid + answer
}
emit('updated', result, room=room)
```

新：
```python
result = {
    'msg': f' {user} 更新資料',
    'user': user,
    'room': room,
    'uid': question_uid,             # ← top-level uid 是 question_uid（語意明確）
    'question_uid': question_uid,    # ← 同上，多一個明確命名讓 FE 不用 fallback
    'data': answer_patch,            # ← 純 answer body，跟 Billows 一致
}
emit('updated', result, room=room)
```

**Breaking change**：FE 必須對應改 `socket.on('updated')` handler。Phase 5 Task 5.4 處理。

### Steps

- [ ] **Step 1**: Modify `on_update` per spec.
- [ ] **Step 2**: Confirm `question_uid` and `answer_patch` are still in scope at the broadcast site (they're declared at top of method).
- [ ] **Step 3**: Commit：
  ```
  fix(task-survey): socket on_update broadcast pure answer body in 'data'

  之前 broadcast 的 'data' 是整個 receive_data dict（含 question_uid + answer + uid + user），
  FE 需要 inner.answer 再剝一層。改成跟 Billows 一致：
  - 'data' 直接是 answer body
  - 'question_uid' top-level 提供（FE 不用 fallback 到 'uid'）

  Breaking change for FE socket.on('updated') handler — Phase 5 同步改。

  對應 plan: ... (Phase 4 Task 4.4)
  ```

---

## Task 4.5：BE service-level tests

**Files:**
- Modify: `tests/test_task_survey_answer.py`

### Spec — 加 5 個 tests

```python
def test_checkpoint_applies_patches_only(
    question_answer_service, seeded_task_survey, cmmgr_engine, app
):
    """Checkpoint 套 dirty patches，只動 patches 提到的題目；其他題的舊值不變動。"""
    ts_uid = seeded_task_survey["uid"]
    ts_id = seeded_task_survey["id"]
    q_uid_1 = seeded_task_survey["question_uids"][0]
    q_uid_2 = seeded_task_survey["question_uids"][1]

    # 先 round 1：兩題都填 v1
    with app.app_context():
        question_answer_service.update_task_survey_answer(
            ts_uid, 1,
            {q_uid_1: {"answer": "v1-1"}, q_uid_2: {"answer": "v1-2"}},
            "test-user",
        )

    # round 2：checkpoint 只送 q1 的 patch，q2 不送
    with app.app_context():
        question_answer_service.checkpoint_task_survey_answer(
            ts_uid, 1,
            [{"question_uid": q_uid_1, "answer": {"answer": "v2-1"}}],
            "test-user",
        )

    with cmmgr_engine.connect() as conn:
        rows = dict(conn.execute(
            text("""SELECT sq.uid, qa.answer FROM survey.question_answers qa
                    JOIN survey.survey_questions sq ON sq.id = qa.question_id
                    WHERE qa.task_survey_id = :tid"""),
            {"tid": ts_id},
        ).fetchall())
    assert rows[q_uid_1] == "v2-1"  # q1 改了
    assert rows[q_uid_2] == "v1-2"  # q2 沒被覆蓋（重點）


def test_checkpoint_fills_empty_rows_for_missing_questions(
    question_answer_service, seeded_task_survey, cmmgr_engine, app
):
    """Checkpoint 對 survey 有題目但 DB 沒 row 的題目，補一筆空 row。"""
    ts_uid = seeded_task_survey["uid"]
    ts_id = seeded_task_survey["id"]
    q_uid_1 = seeded_task_survey["question_uids"][0]

    # checkpoint 只 patch q1，q2 從未被填過
    with app.app_context():
        question_answer_service.checkpoint_task_survey_answer(
            ts_uid, 1,
            [{"question_uid": q_uid_1, "answer": {"answer": "filled"}}],
            "test-user",
        )

    with cmmgr_engine.connect() as conn:
        cnt = conn.execute(
            text("SELECT count(*) FROM survey.question_answers WHERE task_survey_id = :tid"),
            {"tid": ts_id},
        ).scalar()
    # seeded_task_survey 種了 2 題，checkpoint 應該補滿到 2 row（q1 + q2 空 row）
    assert cnt == 2


def test_checkpoint_writes_history_once(
    question_answer_service, seeded_task_survey, cmmgr_engine, app
):
    """Checkpoint 一次呼叫只寫一筆 history（main）。"""
    ts_uid = seeded_task_survey["uid"]
    ts_id = seeded_task_survey["id"]
    q_uid_1 = seeded_task_survey["question_uids"][0]

    with app.app_context():
        question_answer_service.checkpoint_task_survey_answer(
            ts_uid, 1,
            [{"question_uid": q_uid_1, "answer": {"answer": "v1"}}],
            "test-user",
        )

    with cmmgr_engine.connect() as conn:
        h_cnt = conn.execute(
            text("SELECT count(*) FROM survey.question_answer_histories WHERE task_survey_id = :tid"),
            {"tid": ts_id},
        ).scalar()
    assert h_cnt == 1


def test_checkpoint_updates_status(
    question_answer_service, seeded_task_survey, cmmgr_engine, app
):
    """Checkpoint 帶 status 應更新 task_survey.status。"""
    ts_uid = seeded_task_survey["uid"]
    ts_id = seeded_task_survey["id"]
    q_uid_1 = seeded_task_survey["question_uids"][0]

    with app.app_context():
        question_answer_service.checkpoint_task_survey_answer(
            ts_uid, 2,  # 提交
            [{"question_uid": q_uid_1, "answer": {"answer": "submitted"}}],
            "test-user",
        )

    with cmmgr_engine.connect() as conn:
        status = conn.execute(
            text("SELECT status FROM survey.task_surveys WHERE id = :tid"),
            {"tid": ts_id},
        ).scalar()
    assert status == 2


def test_checkpoint_100_questions_select_count_within_budget(
    question_answer_service, seeded_task_survey_100q, app
):
    """100 題 checkpoint 的 SELECT 計數合理（防 N+1 regression）。"""
    from sqlalchemy import event
    ts_uid = seeded_task_survey_100q["uid"]

    # 先填 100 題
    answers_round1 = {
        q_uid: {"answer": f"v_{i}"}
        for i, q_uid in enumerate(seeded_task_survey_100q["question_uids"])
    }
    with app.app_context():
        question_answer_service.update_task_survey_answer(ts_uid, 1, answers_round1, "test-user")

    counts = {"select": 0}
    def _on_cursor_execute(conn, cursor, statement, parameters, context, executemany):
        s = statement.lower()
        if s.startswith("select") and "survey.question_answers" in s:
            counts["select"] += 1

    engine = app.db_engine
    patches = [
        {"question_uid": q_uid, "answer": {"answer": f"v2_{i}"}}
        for i, q_uid in enumerate(seeded_task_survey_100q["question_uids"])
    ]

    counts["select"] = 0
    event.listen(engine, "before_cursor_execute", _on_cursor_execute)
    try:
        with app.app_context():
            question_answer_service.checkpoint_task_survey_answer(
                ts_uid, 1, patches, "test-user"
            )
    finally:
        event.remove(engine, "before_cursor_execute", _on_cursor_execute)

    # 100 題 checkpoint：1 prefetch + 100 in_memory update（各 1 SELECT 在 repo.update 內 SELECT-by-uid）
    # 預期 ≤ 200 SELECT；舊 update_task_survey_answer 是 ≤ 201 範圍
    print(f"[checkpoint select-count] 100 questions: {counts['select']} SELECTs")
    assert counts["select"] <= 220, (
        f"checkpoint 100 questions used {counts['select']} SELECTs against survey.question_answers, "
        "expected ≤ 220 (regression check)"
    )
```

### Steps

- [ ] **Step 1**: 加 5 個 tests 到既有 test 檔末尾。
- [ ] **Step 2**: 跑 `.venv/bin/pytest tests/test_task_survey_answer.py -v`：**所有測試（既有 11 + 新 5 = 16）必須 PASS**。
- [ ] **Step 3**: 連跑 5 次 idempotent。
- [ ] **Step 4**: Commit：
  ```
  test(task-survey): cover checkpoint endpoint behavior

  - patches-only update（其他題不被覆蓋）
  - 補空 row for missing questions
  - 寫一筆 history per call
  - 更新 status
  - 100 題 SELECT 計數防 regression

  對應 plan: ... (Phase 4 Task 4.5)
  ```

---

## Task 4.6：寫 Phase 4 BE changelog

**Files:**
- Create: `docs/changelog/2026-04-30-tweak-task-survey-checkpoint-endpoint.md`

### Spec

```markdown
---
type: tweak
breaking: true
modules: [task_survey]
issue: docs/issues/resolved/2026-04-28-task-survey-answer-duplicate.md
commit: <最後一筆>
---

# tweak(task-survey): 暫存/提交改 patches-based checkpoint endpoint + socket broadcast shape 對齊

## 為什麼

舊 PUT `/project-survey/answers/<uid>` 暫存/提交每次送整份 `answers` dict，
100 題問卷每次儲存都重寫 100 row + 1 history snapshot。多人協作下 DB 寫入塞車。

對應 Billows 已 production 驗證的設計：暫存/提交只送 dirty patches，BE 套 patches +
補空 row（保留全題目都有 row 的 revert 語意）+ 寫一筆 history。

順帶把 socket on_update broadcast 的 shape 跟 Billows 對齊：
舊版 `'data': receive_data`（整 wrapping dict），FE 要 `inner.answer` 再剝一層。
新版 `'data': answer_patch`（純 answer body）+ top-level `'question_uid'`，FE 直接拿。

## 變更

[完整 list 從 plan 的 task 1-5 對應的檔案 / commits]

## API 變更

### 新增
- `POST /api/1.0/project-survey/answers/<uid>/checkpoint`
  body `{status: int, patches: [{question_uid, answer: {answer/ext_answer/feedback/ext_feedback/score}}]}`
  response `{status: true, data: <TaskSurveyDTO>}`

### 廢棄但保留（Excel 批次匯入仍用）
- `PUT /api/1.0/project-survey/answers/<uid>`
  body `{status, answers: {q_uid: {...}}}`

### Socket（breaking — FE 同步改）
- `on 'updated'` broadcast：
  - 舊：`{user, room, uid: q_uid, data: <full receive_data dict>}`
  - 新：`{user, room, uid: q_uid, question_uid: q_uid, data: <answer body>}`

## 測試結果

[16 個 tests pass，5 連跑無 flake，100 題 checkpoint 在 SELECT budget 內]

## 上線清單

1. BE 部署本 commit 之後
2. FE 必須跟著上 Phase 5 commits（前後端綁版本，否則 socket broadcast shape 對不上）
3. 不需要 DB migration

## 提交清單

[列 4.1 ~ 4.5 commits]
```

### Steps

- [ ] **Step 1**: 寫 changelog 檔。
- [ ] **Step 2**: 補各 task 的 commit hash 到表格。
- [ ] **Step 3**: Commit：
  ```
  docs(changelog): task-survey checkpoint endpoint phase 4

  對應 plan: ... (Phase 4 Task 4.6)
  ```

---

# Phase 5 — FE: setAnswer reactivity 修正 + 切到 checkpoint endpoint

**Outcome:**
- bug #1 解決（user1 → user2 → user1 修改不會再卡住）
- 暫存/提交只送 dirty patches、不送整份
- socket broadcast shape 對齊 BE 新 spec
- joined snapshot 不蓋 dirty
- FE changelog 寫好

## Task 5.1：API 常數

**Files:**
- Modify: `compliance-manager-fe/src/config/api/api.js`

### Spec

```js
// 與 PROJECT_SURVEY_ANSWERS_PATCH 共用 base，後綴自己拼
PROJECT_SURVEY_ANSWERS_CHECKPOINT: getUrl('/project-survey/answers'),
// 用法：`${PROJECT_SURVEY_ANSWERS_CHECKPOINT}/${task_survey_uid}/checkpoint`
```

### Steps

- [ ] Edit + commit `feat(survey-fill): add PROJECT_SURVEY_ANSWERS_CHECKPOINT constant`

---

## Task 5.2：`setAnswer` / `setExtAnswer` 改 full 物件重新指派（修 bug #1）

**Files:**
- Modify: `compliance-manager-fe/src/views/survey-v2/SurveyPreview.vue`

### Spec

舊：
```js
function setAnswer(questionId, value) {
    if (!answers.value[questionId]) {
        answers.value[questionId] = { answer: value, ext_answer: '', feedback: '', ext_feedback: '', score: null }
    } else {
        answers.value[questionId].answer = value   // ← mutation 在 socket 覆蓋成 plain object 後不 trigger reactive
    }
}
```

新：
```js
function setAnswer(questionId, value) {
    const current = answers.value[questionId]
    answers.value[questionId] = current && typeof current === 'object'
        ? { ...current, answer: value }
        : { answer: value, ext_answer: '', feedback: '', ext_feedback: '', score: null }
}
```

如有 `setExtAnswer` 也同樣處理。如果 `setExtAnswer(qid, optId, value)`：
```js
function setExtAnswer(questionId, optId, value) {
    const current = answers.value[questionId] || { answer: null, ext_answer: {}, feedback: '', ext_feedback: '', score: null }
    const newExt = { ...(current.ext_answer || {}), [optId]: value }
    answers.value[questionId] = { ...current, ext_answer: newExt }
}
```

具體看現有實作 — grep 確認 setExtAnswer / getExtAnswer 邏輯。

### Steps

- [ ] **Step 1**: grep 出所有 setAnswer / setExtAnswer 簽名與用法。
- [ ] **Step 2**: 改 setter，確保所有 path 都走 full 重新指派。
- [ ] **Step 3**: Manual smoke：兩瀏覽器 user1 → user2 → user1 → user2 連續編輯 q1 文字 / dropdown，每次都該成功傳到對方畫面。
- [ ] **Step 4**: Commit：
  ```
  fix(survey-fill): setAnswer / setExtAnswer use full object reassignment (Billows pattern)

  Property mutation (`answers.value[qid].answer = value`) on a Vue reactive proxy
  works fine for in-component state, but stops triggering reactivity after the
  socket 'updated' handler replaces the value with a socket.io-deserialized
  plain object. Subsequent mutations on the plain object don't notify Vue's
  watchers → template doesn't re-render → user sees their edit doesn't sync.

  Fix: always reassign the full object via spread, so the parent reactive
  proxy's set trap fires every time. Mirrors Billows fe EvaluationView.vue:468.

  Resolves the "user1 inputs → user2 updates → no further edits propagate" bug.

  對應 plan: ... (Phase 5 Task 5.2)
  ```

---

## Task 5.3：`dirtyAnswers` 改成「所有未 checkpoint 的改動」

**Files:**
- Modify: `compliance-manager-fe/src/views/survey-v2/SurveyPreview.vue`

### Spec

之前 dirtyAnswers 只追蹤未 blur 的 text 輸入，dropdown / radio / checkbox / blur 都會 `clearDirty`。新邏輯：dirty 留住所有改動，**只在** `flushDirty` 成功 / `checkpoint` 成功才清。

具體變動：

1. **`markDirty` 改存 cloned snapshot**（避免 reference 與 reactive proxy 同步問題）：
   ```js
   function markDirty(questionUid) {
       if (!questionUid) return
       const cur = answers.value[questionUid]
       dirtyAnswers.value.set(questionUid, cur ? { ...cur } : cur)
   }
   ```

2. **新 helper**：
   ```js
   function buildPatchesFromDirty() {
       const patches = []
       dirtyAnswers.value.forEach((answer, questionUid) => {
           patches.push({ question_uid: questionUid, answer })
       })
       return patches
   }
   ```

3. **template 改寫**：
   - **text/textarea `@input`**：保留 `markDirty(qid)`。
   - **text/textarea `@blur`**：原本 `(e) => { emitEndEdit(qid); clearDirty(qid) }` → 改成 `(e) => { markDirty(qid); emitEndEdit(qid) }`。**移除 clearDirty**。`markDirty` 在 blur 也呼一次（捕捉 @input 後最新的值，包含可能的 `@change` mutations）。
   - **dropdown / radio / checkbox / FileUpload**：原本 `setAnswer(...); emitUpdate(qid); clearDirty(qid)` → 改成 `setAnswer(...); markDirty(qid); emitUpdate(qid)`。**移除 clearDirty**，改加 `markDirty`（在 setAnswer 之後、emitUpdate 之前）。

   `markDirty` 一律在 setAnswer 後立即叫，捕捉最新值。

4. **`flushDirty` 內**：成功時清 snapshot 對應的 dirty keys（避免清掉期間新進的 dirty）— 跟 Billows 設計一樣（snapshot Map + selective clear）：
   ```js
   async function flushDirty({ awaitable = false } = {}) {
       if (!taskSurveyUid.value || dirtyAnswers.value.size === 0) return
       const snapshot = new Map(dirtyAnswers.value)
       const patches = []
       snapshot.forEach((answer, questionUid) => {
           patches.push({ question_uid: questionUid, answer })
       })
       const url = `${API.PROJECT_SURVEY_ANSWERS_PATCH}/${taskSurveyUid.value}/patch`
       const options = {
           method: 'POST',
           body: JSON.stringify({ patches }),
           headers: {
               'Content-Type': 'application/json',
               'Authorization': `Bearer ${accessToken.value}`,
           },
       }
       const clearSnapshotFromDirty = () => {
           snapshot.forEach((_, questionUid) => {
               if (dirtyAnswers.value.get(questionUid) === snapshot.get(questionUid)) {
                   dirtyAnswers.value.delete(questionUid)
               }
           })
       }
       if (awaitable) {
           try {
               const res = await fetch(url, options)
               if (res.ok) clearSnapshotFromDirty()
           } catch {}
       } else {
           options.keepalive = true
           try { fetch(url, options) } catch {}
           clearSnapshotFromDirty()
       }
   }
   ```

### Steps

- [ ] Implement per spec, manual smoke text 打字 + dropdown 切換、確認 dirty Map 累積、暫存後 dirty 清空。
- [ ] Commit `fix(survey-fill): dirtyAnswers tracks all uncommitted changes`

---

## Task 5.4：暫存/提交改打 `/checkpoint` + dirty patches

**Files:**
- Modify: `compliance-manager-fe/src/views/survey-v2/SurveyPreview.vue`

### Spec

舊：
```js
async function onTempSave() {
    await baseService.put(`${API.PROJECT_SURVEY_ANSWERS}/${taskSurveyUid.value}`, {
        status: 1,
        answers: answers.value,
    })
}
async function onSave(status) {
    ... validate ...
    await baseService.put(`${API.PROJECT_SURVEY_ANSWERS}/${taskSurveyUid.value}`, {
        status,
        answers: payload,
    })
    await loadTaskSurvey(taskSurveyUid.value)
}
```

新：
```js
async function onTempSave() {
    if (!taskSurveyUid.value) return
    submitted.value = true
    try {
        const patches = buildPatchesFromDirty()
        await baseService.post(
            `${API.PROJECT_SURVEY_ANSWERS_CHECKPOINT}/${taskSurveyUid.value}/checkpoint`,
            { status: 1, patches },
        )
        dirtyAnswers.value.clear()
        toast.add({ severity: 'success', summary: t('lang.survey_v2.preview.save_success'), life: 3000 })
    } catch (e) {
        toast.add({ severity: 'error', summary: t('lang.common.error'), detail: e?.message || '', life: 5000 })
    } finally {
        submitted.value = false
    }
}

async function onSave(status) {
    if (!taskSurveyUid.value) {
        toast.add({ severity: 'success', summary: t('lang.survey_v2.preview.submit_success') + '（預覽模式）', life: 3000 })
        return
    }
    // validate required fields for status 2 / 9 (same logic as before)
    if (status === 2 || status === 9) {
        const missing = allQuestions.value.filter(q => q.isRequired && !isAnswered(q.id))
        if (missing.length > 0) {
            const names = missing.map(q => q.questionNumber || q.question).join('、')
            toast.add({ severity: 'error', summary: t('lang.survey_v2.preview.required_fields_missing'), detail: names, life: 8000 })
            return
        }
    }
    submitted.value = true
    try {
        const patches = buildPatchesFromDirty()
        // score 整數化（沿用既有處理）
        for (const p of patches) {
            if (p.answer && p.answer.score != null) {
                p.answer.score = parseInt(p.answer.score, 10)
            }
        }
        await baseService.post(
            `${API.PROJECT_SURVEY_ANSWERS_CHECKPOINT}/${taskSurveyUid.value}/checkpoint`,
            { status, patches },
        )
        dirtyAnswers.value.clear()
        const msgMap = {
            2: t('lang.survey_v2.preview.submit_success'),
            3: t('lang.survey_v2.preview.review_success'),
            9: t('lang.survey_v2.preview.close_success'),
        }
        toast.add({ severity: 'success', summary: msgMap[status] || t('lang.survey_v2.preview.submit_success'), life: 3000 })
        await loadTaskSurvey(taskSurveyUid.value)
    } catch (e) {
        toast.add({ severity: 'error', summary: t('lang.common.error'), detail: e?.message || '', life: 5000 })
    } finally {
        submitted.value = false
    }
}
```

### Steps

- [ ] Implement per spec.
- [ ] Manual smoke：暫存 → BE log 確認收到 patches；按提交 → status=2 寫入 + 必填驗證仍 work。
- [ ] Commit `fix(survey-fill): tempSave/submit POST to /checkpoint with dirty patches`

---

## Task 5.5：`socket.on('updated')` 對齊新 broadcast shape + dirty 防覆蓋

**Files:**
- Modify: `compliance-manager-fe/src/views/survey-v2/composables/useSurveySocket.js`

### Spec

useSurveySocket 需要從 caller 收一個 `isDirty(questionUid)` callback：

```js
export function useSurveySocket({
    taskSurveyUid,
    userLoginName,
    answers,
    onAnswersReload,
    onUserJoined,
    onUserLeft,
    isDirty = () => false,   // ← 新增 optional callback
}) {
    ...
}
```

`socket.on('updated')` 改：
```js
socket.on('updated', (data) => {
    if (data.user === userLoginName.value) return
    // BE Phase 4: 'data' 直接是 answer body；'question_uid' top-level 提供
    const questionUid = data.question_uid || data.uid
    const answerBody = data.data
    if (!questionUid || answerBody === undefined) return
    // 使用者本地對該題還在改 → 不覆蓋（避免回灌洗掉本地未送的內容）
    if (isDirty(questionUid)) return
    answers.value[questionUid] = answerBody
})
```

`socket.on('joined')` 改 dirty-aware merge：
```js
socket.on('joined', (data) => {
    onlineUsers.value = data.users || []
    if (data.user === userLoginName.value) {
        if (data.data) {
            const merged = { ...answers.value }
            Object.keys(data.data).forEach((qid) => {
                if (!isDirty(qid)) {
                    merged[qid] = data.data[qid]
                }
            })
            answers.value = merged
        }
    } else {
        onUserJoined?.(data)
    }
})
```

SurveyPreview.vue 傳 `isDirty`：
```js
const { ... } = useSurveySocket({
    taskSurveyUid,
    userLoginName: computed(() => userInfo.value?.login_name),
    answers,
    onAnswersReload: () => loadTaskAnswers(taskSurveyUid.value),
    onUserJoined: ...,
    onUserLeft: ...,
    isDirty: (qid) => dirtyAnswers.value.has(qid),
})
```

### Steps

- [ ] Implement per spec, both files.
- [ ] Manual smoke 兩瀏覽器：
  - user2 廣播 q1 → user1 dirty 沒有該 key → 應收到 update。
  - user1 dirty 有 q1（正在打字）→ 應**不**收到 update（不被洗掉）。
  - user1 進房，user2 已填 q3 → user1 看到 q3 更新；但 user1 已 dirty 的題不被覆蓋。
- [ ] Commit `fix(survey-fill): socket on 'updated'/'joined' align with BE phase 4 + dirty-aware merge`

---

## Task 5.6：寫 Phase 5 FE changelog

**Files:**
- Create: `compliance-manager-fe/docs/changelog/survey-fill-collab-fix/20260430_<HHMM>_phase5-checkpoint-changelog.md`

內容：對應 Phase 5 全部變動，含：bug #1 fix、checkpoint endpoint 切換、socket shape 對齊、dirty-aware merge。手動 smoke 5 情境。Staging / prod 上線必須與 BE Phase 4 同版本。

- [ ] Write + commit。

---

# Phase 6 — 收尾

## Task 6.1：跑全套測試 + manual smoke

- [ ] BE pytest 全綠（16 tests，5 連跑無 flake）
- [ ] FE Vue compile 無 error
- [ ] 兩瀏覽器 manual smoke 全 5 情境綠（含本次特別關注的 user1 → user2 → user1 連續編輯）
- [ ] Dev DB 跑：
  ```sql
  SELECT COUNT(*) FROM (SELECT 1 FROM survey.question_answers GROUP BY task_survey_id, question_id HAVING COUNT(*)>1) g;
  -- 應為 0（先前 Phase 1 修好，這次不 regression）
  ```

## Task 6.2：在 issue resolved/2026-04-28 追加 follow-up

加段落「Phase 4-5 進階優化（2026-04-30）」描述 checkpoint endpoint 上線、broadcast shape 對齊、setAnswer reactivity 修正。

## Task 6.3：Staging / Prod 上線清單

寫在最終回覆裡：
1. 兩個 repo 對應 commit hash
2. 部署順序 BE 先 FE 後（**前後端必須同版本**，因 socket broadcast shape 為 breaking change）
3. 不需要 DB migration
4. 不需要 Redis 操作
5. 既有 PUT `/project-survey/answers/<uid>` 仍然可用（Excel 批次匯入），不會 404

---

## 執行順序與風險

| Phase | Task | 風險 | 緩解 |
|-------|------|------|------|
| 4 | 4.1 service | 低 — 沒動既有 method | service-level pytest |
| 4 | 4.2-4.3 serializer + route | 低 | url_map 驗證 + integration test |
| 4 | 4.4 broadcast shape | **中 — breaking change** | FE Phase 5 同步上、不單獨部署 |
| 4 | 4.5 tests | 低 | pytest 自動化 |
| 4 | 4.6 changelog | 低 | — |
| 5 | 5.1-5.2 setter fix | 低 — 修 bug、無外部影響 | manual smoke |
| 5 | 5.3 dirty refactor | 中 — 改變多處 template | manual smoke 測各 input 類型 |
| 5 | 5.4 checkpoint switch | 中 — 暫存/提交主流程改變 | 既有 PUT 還在，可一鍵切回（git revert） |
| 5 | 5.5 socket dirty-aware | 中 — 改 useSurveySocket 簽名 | TypeScript 沒有但 JSDoc 有，看 lint |
| 5 | 5.6 changelog | 低 | — |

**rollback 策略：** 前後端綁版本同步 revert。BE Phase 4 不單獨 rollback（FE 會壞）；FE Phase 5 不單獨 rollback（暫存按鈕會 404 因為 `/checkpoint` 在但前端找不到）。同時 revert 才一致。

## 預估時間

約 4-6 小時（包含 task 派發 + spec 與 code review + commits）。

---

## Plan Review 後決定執行方式

依前一輪 Phase 1-3 的經驗，建議：
- **subagent-driven**，每個 task 派 fresh subagent + 雙 review
- 6 BE tasks (4.1-4.6) + 6 FE tasks (5.1-5.6) + 3 收尾 (6.1-6.3) ≈ 15 tasks
- 預期 13-15 個 commits 在 BE，6-8 個在 FE

執行中重要決策點停下來確認（例：4.4 broadcast shape 上線前；5.4 暫存切換前的 manual smoke）。