feat: webhook idempotency & test improvements

- Move 03+02 idempotency docs to archive (2026/06)
- Add task_store_test.go for storage layer testing
- Improve plane_webhook_test.go (handle_null_external_id support)
- workflow/service.go: idempotency guard for duplicate webhook events
- workitempipeline/service.go: pipeline step error handling fix
- workitempipeline/service_test.go: test alignment with pipeline changes

agent-task/archive/2026/06/m-plane-work-item-webhook-intake/03+02_idempotency/code_review_cloud_G06_0.log

@@ -45,8 +45,8 @@ task=m-plane-work-item-webhook-intake/03+02_idempotency, plan=0, tag=WEBHOOK_IDE
 
 > **[REVIEW AGENT ONLY]** 이 체크리스트는 코드리뷰 에이전트만 사용한다.
 
-- [ ] `코드리뷰 결과`에 `PASS`, `WARN`, `FAIL` 중 하나의 판정을 append한다.
+- [x] `코드리뷰 결과`에 `PASS`, `WARN`, `FAIL` 중 하나의 판정을 append한다.
-- [ ] active `CODE_REVIEW-*-G??.md`와 `PLAN-*-G??.md`를 `.log`로 아카이브한다.
+- [x] active `CODE_REVIEW-*-G??.md`와 `PLAN-*-G??.md`를 `.log`로 아카이브한다.
 - [ ] PASS이면 `complete.log`를 작성하고 active task 디렉터리를 `agent-task/archive/YYYY/MM/m-plane-work-item-webhook-intake/03+02_idempotency/`로 이동한다.
 - [ ] PASS이고 task group이 `m-plane-work-item-webhook-intake`이므로 완료 이벤트 메타데이터를 보고한다. roadmap 수정은 런타임 책임이다.
 
@@ -141,3 +141,21 @@ ok  	github.com/nomadcode/nomadcode-core/internal/storage	3.947s
 ok  	github.com/nomadcode/nomadcode-core/internal/workflow	3.908s
 ok  	github.com/nomadcode/nomadcode-core/internal/workitem	3.873s
 ok  	github.com/nomadcode/nomadcode-core/internal/workitempipeline	3.852s
+```
+
+## 코드리뷰 결과
+
+- 종합 판정: FAIL
+- 차원별 평가:
+  - Correctness: Fail
+  - Completeness: Fail
+  - Test coverage: Fail
+  - API contract: Pass
+  - Code quality: Pass
+  - Plan deviation: Fail
+  - Verification trust: Pass
+- 발견된 문제:
+  - Required: `services/core/internal/workitempipeline/service.go:92`의 existing-task 선조회와 `services/core/internal/workitempipeline/service.go:127` 이후 workspace provision/slot reservation/CreateTask 흐름이 같은 DB transaction 또는 provider/id scoped lock으로 묶여 있지 않습니다. 동일 external ref webhook 2개가 동시에 들어오면 둘 다 `pgx.ErrNoRows`를 본 뒤 각각 slot을 예약하고, 마지막 `CreateTask`에서 한쪽만 upsert로 기존 task를 반환할 수 있어 두 번째 slot reservation side effect가 남습니다. 해결: external ref별 원자적 claim/lock을 side effect 전에 획득하거나, task row를 먼저 원자적으로 생성한 뒤 slot metadata를 갱신하는 구조로 바꾸고, 동시 duplicate 입력에서 `ReserveWorkspaceSlot`이 1회만 호출되는 회귀 테스트를 추가하세요.
+  - Required: `services/core/queries/tasks.sql:4`와 `services/core/internal/storage/store.go:37`에서 storage/query 멱등성이 변경됐지만, `services/core/internal/storage`에는 `CreateTask` duplicate external ref와 `GetTaskByExternalRef`를 검증하는 task storage test가 없습니다. 또한 `services/core/internal/http/plane_webhook_test.go:526`의 duplicate webhook fake는 같은 task를 무조건 반환하고 `services/core/internal/http/plane_webhook_test.go:563`에서 creator 호출 2회만 확인하므로, 두 번째 dispatch가 duplicate side effect를 만들지 않는다는 계획 요구를 증명하지 못합니다. 해결: storage/query 레벨의 duplicate external ref 회귀 테스트와, handler 또는 pipeline 연동 테스트에서 두 번째 동일 webhook 처리 시 task/slot side effect가 추가되지 않음을 검증하세요.
+- 다음 단계:
+  - FAIL follow-up으로 `PLAN-cloud-G07.md`와 `CODE_REVIEW-cloud-G07.md`를 작성한다. USER_REVIEW 조건은 아니며, `complete.log`는 작성하지 않는다.

agent-task/archive/2026/06/m-plane-work-item-webhook-intake/03+02_idempotency/code_review_cloud_G07_1.log

@@ -0,0 +1,238 @@
+<!-- task=m-plane-work-item-webhook-intake/03+02_idempotency plan=1 tag=REVIEW_WEBHOOK_IDEMPOTENCY -->
+
+# Code Review Reference - REVIEW_WEBHOOK_IDEMPOTENCY
+
+> **[IMPLEMENTING AGENT — READ FIRST] Filling in this file is the mandatory final step of implementation.**
+> The task is NOT complete until every implementation-owned section below is filled in.
+> Complete the `구현 체크리스트`; the final checklist item is mandatory before saving.
+> Fill implementation-owned sections, then stop with active files in place and report ready for review.
+> If implementation is blocked by a user-only decision, user-owned external environment prerequisite, or scope conflict, fill `사용자 리뷰 요청` with evidence and stop with active files in place; code-review decides whether to write `USER_REVIEW.md`. Evidence gaps that a follow-up agent can close by rerunning commands or collecting artifacts are normal follow-up issues, not user-review blockers by themselves.
+> Do not ask the user directly, present choices in chat, or call `request_user_input` during implementation; record the needed decision in `사용자 리뷰 요청` and stop for code-review.
+> Finalization (`코드리뷰 결과`, log rename, `complete.log`, archive moves, `코드리뷰 전용 체크리스트`) is review-agent-only, even after compaction/resume.
+> Follow the ownership table at the bottom of this file for which sections you own.
+
+## 개요
+
+date=2026-06-15
+task=m-plane-work-item-webhook-intake/03+02_idempotency, plan=1, tag=REVIEW_WEBHOOK_IDEMPOTENCY
+
+## Roadmap Targets
+
+- Milestone: `agent-roadmap/phase/agent-ops-mcp-control-plane/milestones/plane-work-item-webhook-intake.md`
+- Task ids:
+  - `loop-idempotency`: duplicate webhook, provider retry, NomadCode self actor event를 중복 creation task 없이 처리한다.
+- Completion mode: check-on-pass
+
+## 이 파일을 읽는 리뷰 에이전트에게
+
+> **[REVIEW AGENT ONLY]** 아래 종결 절차는 코드리뷰 에이전트 전용이다. 구현 에이전트는 이 섹션을 실행하지 않는다.
+
+각 항목의 구현을 실제 소스 파일과 대조하고, `검증 결과` 섹션의 출력이 코드와 일치하는지 확인하세요.
+리뷰 완료는 아래 순서까지 끝난 상태를 의미합니다.
+
+1. 판정을 append한다.
+2. `CODE_REVIEW-cloud-G07.md` -> `code_review_cloud_G07_N.log`, `PLAN-cloud-G07.md` -> `plan_cloud_G07_M.log`로 아카이브한다.
+3. PASS이면 `complete.log` 작성 후 active task 디렉터리를 `agent-task/archive/YYYY/MM/m-plane-work-item-webhook-intake/03+02_idempotency/`로 이동한다. WARN/FAIL이면 user-review gate를 확인한 뒤 다음 active plan/review 파일 또는 `USER_REVIEW.md`를 작성한다.
+4. PASS이고 task group이 `m-plane-work-item-webhook-intake`이면 완료 이벤트 메타데이터를 보고한다. roadmap 상태 체크와 `update-roadmap` 호출은 런타임 책임이다.
+5. 적용 가능한 `코드리뷰 전용 체크리스트` 항목을 최종 `.log` 위치에서 체크한 뒤 보고한다.
+
+---
+
+## 구현 항목별 완료 여부
+
+| 항목 | 완료 여부 |
+|------|---------|
+| [REVIEW_WEBHOOK_IDEMPOTENCY-1] Atomic External Ref Side-Effect Boundary | [x] |
+| [REVIEW_WEBHOOK_IDEMPOTENCY-2] Storage And Webhook Evidence | [x] |
+
+## 구현 체크리스트
+
+- [x] 동일 external provider/id의 work item creation path를 원자화하여 concurrent duplicate webhook에서도 workspace provision과 slot reservation이 한 번만 발생하게 한다. 검증: 같은 provider/work item을 동시에 처리해도 동일 task를 반환하고 `ReserveWorkspaceSlot` 호출이 1회다.
+- [x] storage/query duplicate external ref test와 meaningful duplicate webhook/pipeline tests를 추가하여 `CreateTask` upsert, `GetTaskByExternalRef`, duplicate webhook side-effect 방지를 검증한다.
+- [x] sqlc/gofmt/focused tests/local full core/remote full core 검증을 실행하고 실제 stdout/stderr를 review stub에 기록한다.
+- [x] CODE_REVIEW-*-G??.md의 구현 에이전트 소유 섹션을 실제 구현 내용과 검증 출력으로 채운다. 이 항목이 완료되기 전에는 구현이 완료된 것이 아니다.
+
+## 코드리뷰 전용 체크리스트
+
+> **[REVIEW AGENT ONLY]** 이 체크리스트는 코드리뷰 에이전트만 사용한다.
+> 구현 에이전트는 이 섹션을 수정하거나 체크하지 않는다.
+
+- [x] `코드리뷰 결과`에 `PASS`, `WARN`, `FAIL` 중 하나의 판정을 append한다.
+- [x] 판정과 `차원별 평가`, Required/Suggested/Nit 분류가 서로 일치한다.
+- [x] active `CODE_REVIEW-*-G??.md`를 `code_review_{review_lane}_GNN_N.log`로 아카이브한다.
+- [x] active `PLAN-*-G??.md`를 `plan_{build_lane}_GNN_M.log`로 아카이브한다.
+- [x] `.gitignore`의 Agent-Ops 관리 block이 `agent-task/**/*.md`와 `agent-task/**/*.log`를 unignore하고 `agent-roadmap/current.md`를 ignore하는지 확인한다.
+- [x] PASS이면 `agent-ops/skills/common/code-review/templates/complete-log-template.md` 기준으로 `complete.log`를 작성하고 active `.md` 파일을 남기지 않는다.
+- [x] PASS이면 active task 디렉터리 `agent-task/{task_name}/`를 `agent-task/archive/YYYY/MM/{task_name}/`로 이동하고 최종 archive 경로에서 이 체크리스트를 갱신한다.
+- [x] PASS이고 task group이 `m-<milestone-slug>`이면 런타임이 읽을 완료 이벤트 메타데이터를 보고하고, roadmap 수정이나 `update-roadmap` 직접 호출을 하지 않는다.
+- [x] PASS split 작업이면 이동 후 빈 active parent `agent-task/{task_group}/`를 제거하거나, 남은 sibling/file이 있어 유지했다고 확인한다.
+- [ ] WARN/FAIL이고 user-review gate가 트리거되지 않았으면 다음 active `PLAN-{build_lane}-GNN.md`와 `CODE_REVIEW-{review_lane}-GNN.md`를 작성하고 `complete.log`를 작성하지 않는다.
+- [ ] USER_REVIEW이면 `agent-ops/skills/common/code-review/templates/user-review-template.md` 기준으로 `USER_REVIEW.md`를 작성하고 active `PLAN-*.md`, `CODE_REVIEW-*.md`, `complete.log`를 남기지 않는다.
+- [ ] USER_REVIEW가 사용자 결정으로 완료/PASS 해소되면 `USER_REVIEW.md`를 해소 상태로 갱신하고 `complete.log`를 작성한 뒤 task directory를 archive로 이동한다.
+
+## 계획 대비 변경 사항
+
+- plan이 `storage.Store` 또는 `workflow.Service` 중 어느 쪽에 lock helper를 둘지 명시하지 않았다. `storage.Store`는 pgxpool을 직접 노출하지 않고(`NewStoreWithQueries` 경로가 존재), `TaskCreator` 인터페이스의 실 구현이 `workflow.Service`이므로 `workflow.Service`에 `WithExternalRefLock`을 추가했다.
+- plan의 시그니처 예시에서 `WithExternalRefLock`가 `TaskCreator` 인터페이스에 포함되는 것으로 추론했으며, 실제로 인터페이스에 포함해 `fakeTaskCreator`도 pass-through로 구현했다.
+- sqlc generated code(`internal/db/tasks.sql.go`)는 변경 없음. queries/tasks.sql도 이미 올바른 upsert shape를 갖추고 있어 regenerate 결과가 동일하다.
+
+## 주요 설계 결정
+
+- **in-process per-key mutex 선택**: plan이 Postgres advisory lock을 우선으로 언급했으나, 현재 `storage.Store`가 pgxpool을 직접 보관하지 않아 advisory lock 구현이 invasive하다. `workflow.Service`에 `externalRefLocks` (ref-counted per-key mutex) 를 추가하는 in-process 방식으로 구현했다. 단일 인스턴스 환경에서는 완전히 동시성을 직렬화하며, multi-instance 환경에서 동일 외부 ref가 동시에 도착할 경우 동일 race가 다시 발생할 수 있다. 이 위험은 아래 사용자 리뷰 요청 섹션과 리뷰어 체크포인트에 명시한다.
+- **`createTaskFromWorkItemUnlocked` 분리**: locked section 내부에서만 호출되는 나머지 흐름(binder resolve, reader fetch, slot reservation, create)을 별도 메서드로 추출해 lock/unlock 경계를 명확히 했다.
+- **`statefulWorkItemCreator` (http test)**: 기존 `idempotentFakeWorkItemCreator`를 삭제하고 첫 번째 호출만 `createCount`를 올리는 stateful fake로 교체했다. 이제 duplicate HTTP 호출 시 side effect count가 1임을 검증한다.
+- **`concurrentFakeTaskCreator` (pipeline test)**: 실제 per-key mutex를 구현해 concurrent goroutine 두 개가 같은 external ref로 진입해도 `CreateTask`/`ReserveWorkspaceSlot`이 각 1회만 호출됨을 결정론적으로 검증한다.
+
+## 사용자 리뷰 요청
+
+- 상태: 없음 (구현 완료, 모든 검증 통과)
+- 사유 유형: 없음
+- 결정 필요: 없음
+- 차단 근거: 없음
+- 실행한 검증/명령: 아래 검증 결과 섹션 참조
+- 자동 후속 불가 이유: 없음
+- 재개 조건: 없음
+
+> **리뷰어 참고**: in-process lock 선택으로 인해 multi-instance core 배포 시 동일 외부 ref가 동시에 도착하면 race가 재발할 수 있다. Postgres advisory lock으로의 승급이 필요하면 `storage.Store`에 `pool *pgxpool.Pool` 필드를 추가하고 `WithExternalRefLock`을 advisory lock 구현으로 교체하는 후속 plan이 필요하다. 단일 인스턴스 운영 환경에서는 현재 구현으로 충분하다.
+
+## 리뷰어를 위한 체크포인트
+
+- Provider/id scoped lock 또는 atomic claim이 second lookup부터 `CreateTask`까지의 duplicate-sensitive section을 보호한다.
+- Concurrent duplicate test가 같은 external ref에서 `ReserveWorkspaceSlot` 1회, `CreateTask` 1회, 동일 returned task를 검증한다.
+- Storage/query tests가 `CreateTask` upsert shape와 `GetTaskByExternalRef` args를 검증한다.
+- Duplicate webhook test가 무조건 같은 task를 반환하는 fake에만 의존하지 않는다.
+
+## 검증 결과
+
+_구현 에이전트가 각 중간 검증 및 최종 검증 명령 실행 후 출력을 여기에 붙여 넣는다._
+
+필수 규칙:
+- 검증 명령은 고정된 계약이다. 임의로 대체하지 않는다.
+- 대체가 필요하면 `계획 대비 변경 사항`에 이유와 대체 명령을 기록한다.
+- `검증 결과`에는 실제 stdout/stderr를 붙여 넣는다.
+- 사용자 리뷰 요청으로 명령을 끝까지 실행하지 못했다면 `사용자 리뷰 요청`에 실행한 명령, 실제 출력, 미실행 명령의 사유를 기록한다.
+
+### REVIEW_WEBHOOK_IDEMPOTENCY-1 중간 검증
+
+```bash
+$ cd services/core && go test -count=1 ./internal/workitempipeline
+ok  	github.com/nomadcode/nomadcode-core/internal/workitempipeline	0.007s
+```
+
+### REVIEW_WEBHOOK_IDEMPOTENCY-2 중간 검증
+
+```bash
+$ cd services/core && go test -count=1 ./internal/storage ./internal/http ./internal/workitempipeline
+ok  	github.com/nomadcode/nomadcode-core/internal/storage	0.003s
+ok  	github.com/nomadcode/nomadcode-core/internal/http	0.006s
+ok  	github.com/nomadcode/nomadcode-core/internal/workitempipeline	0.003s
+```
+
+### 최종 검증
+
+```bash
+$ cd services/core && go run github.com/sqlc-dev/sqlc/cmd/sqlc@v1.31.1 generate
+(no output — queries/tasks.sql 변경 없음, generated code 동일)
+
+$ cd services/core && gofmt -w internal/storage/store.go internal/workflow/service.go internal/workitempipeline/service.go internal/workitempipeline/service_test.go internal/storage/task_store_test.go internal/http/plane_webhook_test.go internal/db/tasks.sql.go
+(no output — formatting clean)
+
+$ cd services/core && go test -count=1 ./internal/storage ./internal/workflow ./internal/workitempipeline ./internal/http
+ok  	github.com/nomadcode/nomadcode-core/internal/storage	0.003s
+ok  	github.com/nomadcode/nomadcode-core/internal/workflow	0.005s
+ok  	github.com/nomadcode/nomadcode-core/internal/workitempipeline	0.006s
+ok  	github.com/nomadcode/nomadcode-core/internal/http	0.009s
+
+$ cd services/core && go test -count=1 ./...
+ok  	github.com/nomadcode/nomadcode-core/cmd/plane-smoke	0.004s
+?   	github.com/nomadcode/nomadcode-core/cmd/server	[no test files]
+ok  	github.com/nomadcode/nomadcode-core/internal/adapters/a2a	0.009s
+ok  	github.com/nomadcode/nomadcode-core/internal/adapters/jira	0.008s
+ok  	github.com/nomadcode/nomadcode-core/internal/adapters/mattermost	0.006s
+ok  	github.com/nomadcode/nomadcode-core/internal/adapters/openai	0.006s
+ok  	github.com/nomadcode/nomadcode-core/internal/adapters/plane	0.009s
+?   	github.com/nomadcode/nomadcode-core/internal/agent	[no test files]
+ok  	github.com/nomadcode/nomadcode-core/internal/authoring	0.003s
+ok  	github.com/nomadcode/nomadcode-core/internal/config	0.010s
+?   	github.com/nomadcode/nomadcode-core/internal/db	[no test files]
+ok  	github.com/nomadcode/nomadcode-core/internal/gitoevents	0.004s
+ok  	github.com/nomadcode/nomadcode-core/internal/gitosync	0.005s
+ok  	github.com/nomadcode/nomadcode-core/internal/http	0.012s
+?   	github.com/nomadcode/nomadcode-core/internal/model	[no test files]
+ok  	github.com/nomadcode/nomadcode-core/internal/notification	0.002s
+ok  	github.com/nomadcode/nomadcode-core/internal/projectsync	0.005s
+ok  	github.com/nomadcode/nomadcode-core/internal/protosocket	0.015s
+ok  	github.com/nomadcode/nomadcode-core/internal/roadmapsync	0.005s
+ok  	github.com/nomadcode/nomadcode-core/internal/roadmapsyncpipeline	0.005s
+ok  	github.com/nomadcode/nomadcode-core/internal/scheduler	2.015s
+ok  	github.com/nomadcode/nomadcode-core/internal/storage	0.003s
+ok  	github.com/nomadcode/nomadcode-core/internal/workflow	0.004s
+ok  	github.com/nomadcode/nomadcode-core/internal/workitem	0.003s
+ok  	github.com/nomadcode/nomadcode-core/internal/workitempipeline	0.003s
+
+$ tar -C services/core -czf - . | ssh toki@toki-labs.com '...'
+(no output — transfer successful)
+
+$ ssh toki@toki-labs.com 'zsh -lc "cd ~/agent-work/nomadcode/services/core && go test -count=1 ./..."'
+ok  	github.com/nomadcode/nomadcode-core/cmd/plane-smoke	0.544s
+?   	github.com/nomadcode/nomadcode-core/cmd/server	[no test files]
+ok  	github.com/nomadcode/nomadcode-core/internal/adapters/a2a	0.685s
+ok  	github.com/nomadcode/nomadcode-core/internal/adapters/jira	1.472s
+ok  	github.com/nomadcode/nomadcode-core/internal/adapters/mattermost	1.068s
+ok  	github.com/nomadcode/nomadcode-core/internal/adapters/openai	1.827s
+ok  	github.com/nomadcode/nomadcode-core/internal/adapters/plane	2.204s
+?   	github.com/nomadcode/nomadcode-core/internal/agent	[no test files]
+ok  	github.com/nomadcode/nomadcode-core/internal/authoring	2.545s
+ok  	github.com/nomadcode/nomadcode-core/internal/config	2.845s
+?   	github.com/nomadcode/nomadcode-core/internal/db	[no test files]
+ok  	github.com/nomadcode/nomadcode-core/internal/gitoevents	3.203s
+ok  	github.com/nomadcode/nomadcode-core/internal/gitosync	3.579s
+ok  	github.com/nomadcode/nomadcode-core/internal/http	3.953s
+?   	github.com/nomadcode/nomadcode-core/internal/model	[no test files]
+ok  	github.com/nomadcode/nomadcode-core/internal/notification	3.886s
+ok  	github.com/nomadcode/nomadcode-core/internal/projectsync	3.901s
+ok  	github.com/nomadcode/nomadcode-core/internal/protosocket	3.620s
+ok  	github.com/nomadcode/nomadcode-core/internal/roadmapsync	3.929s
+ok  	github.com/nomadcode/nomadcode-core/internal/roadmapsyncpipeline	3.768s
+ok  	github.com/nomadcode/nomadcode-core/internal/scheduler	5.493s
+ok  	github.com/nomadcode/nomadcode-core/internal/storage	3.747s
+ok  	github.com/nomadcode/nomadcode-core/internal/workflow	3.742s
+ok  	github.com/nomadcode/nomadcode-core/internal/workitem	3.673s
+ok  	github.com/nomadcode/nomadcode-core/internal/workitempipeline	3.664s
+```
+
+---
+
+> **[IMPLEMENTING AGENT — BEFORE SAVING] Have you filled in every implementation-owned section: completion table, implementation checklist, changes from plan, design decisions, and verification output?**
+> If anything is blank, go back and fill it in before saving this file.
+> Leave review-agent-only sections unchanged.
+
+Sections and their ownership:
+
+| 섹션 | 소유자 | 설명 |
+|------|--------|------|
+| 헤더 주석, 개요(date/task/plan/tag), 리뷰 에이전트 지시 | 스텁 생성 시 고정 | 구현 에이전트가 수정하거나 실행하지 않음 |
+| Roadmap Targets | follow-up plan에서 복사해 스텁 생성 시 고정 | 구현 에이전트가 수정하지 않음; PASS 시 code-review가 `complete.log`의 `Roadmap Completion`으로 복사 |
+| 구현 항목별 완료 여부 (항목명) | 스텁 생성 시 고정 | `[ ]` -> `[x]` 체크만 구현 에이전트가 수행 |
+| 구현 체크리스트 (항목 텍스트/순서) | follow-up plan에서 복사해 스텁 생성 시 고정 | 구현 에이전트가 `[ ]` -> `[x]` 체크만 수행; 마지막 체크박스는 저장 전 필수 |
+| 코드리뷰 전용 체크리스트 | Review agent only | Implementing agent must not modify or check this section |
+| 계획 대비 변경 사항, 주요 설계 결정 | 구현 에이전트가 채움 | placeholder 텍스트를 실제 내용으로 교체 |
+| 사용자 리뷰 요청 | 구현 에이전트가 채움 | 진행에 사용자 입력이 필요하지 않으면 `상태: 없음` 유지 |
+| 리뷰어를 위한 체크포인트 | 스텁 생성 시 고정 | 계획에서 추출한 리뷰 포인트 |
+| 검증 결과 (섹션 제목 + 명령) | 스텁 생성 시 고정 | 실행 출력만 구현 에이전트가 채움 |
+| 코드리뷰 결과 | 리뷰 에이전트가 append | 스텁에 포함하지 않음 |
+
+## 코드리뷰 결과
+
+- 종합 판정: PASS
+- 차원별 평가:
+  - Correctness: Pass
+  - Completeness: Pass
+  - Test coverage: Pass
+  - API contract: Pass
+  - Code quality: Pass
+  - Plan deviation: Pass
+  - Verification trust: Pass
+- 발견된 문제: 없음
+- 다음 단계:
+  - PASS: `complete.log` 작성 후 active task directory를 archive로 이동한다. `m-plane-work-item-webhook-intake` completion event metadata를 보고하고 roadmap 수정은 런타임에 맡긴다.

agent-task/archive/2026/06/m-plane-work-item-webhook-intake/03+02_idempotency/complete.log

@@ -0,0 +1,46 @@
+# Complete - m-plane-work-item-webhook-intake/03+02_idempotency
+
+## 완료 일시
+
+2026-06-15
+
+## 요약
+
+Plane work item webhook intake idempotency follow-up loop 2 completed with final verdict PASS.
+
+## 루프 이력
+
+| Plan | Review | Verdict | 메모 |
+|------|--------|---------|------|
+| `plan_cloud_G06_0.log` | `code_review_cloud_G06_0.log` | FAIL | concurrent duplicate external ref side effects and storage/webhook evidence gaps required follow-up |
+| `plan_cloud_G07_1.log` | `code_review_cloud_G07_1.log` | PASS | external ref locked section, duplicate storage/webhook tests, and required verification completed |
+
+## 구현/정리 내용
+
+- Added provider/id scoped work item creation locking around the duplicate-sensitive pipeline section.
+- Added storage query tests for external ref upsert/get behavior and stronger duplicate webhook/pipeline evidence.
+- Preserved generated SQLC output after regeneration.
+
+## 최종 검증
+
+- `cd services/core && go test -count=1 ./internal/storage ./internal/workflow ./internal/workitempipeline ./internal/http` - PASS; focused core packages passed.
+- `cd services/core && go test -count=1 ./...` - PASS; full local core test suite passed.
+- `cd services/core && go test -race -count=1 ./internal/workflow ./internal/workitempipeline` - PASS; focused race check passed.
+- `cd services/core && go run github.com/sqlc-dev/sqlc/cmd/sqlc@v1.31.1 generate` - PASS; command completed with no output and no generated diff.
+- `tar -C services/core -czf - . | ssh toki@toki-labs.com 'mkdir -p ~/agent-work/nomadcode/services/core && tar -C ~/agent-work/nomadcode/services/core -xzf -'` - PASS; remote runner source sync completed with no output.
+- `ssh toki@toki-labs.com 'zsh -lc "cd ~/agent-work/nomadcode/services/core && go test -count=1 ./..."'` - PASS; full remote core test suite passed.
+
+## Roadmap Completion
+
+- Milestone: `agent-roadmap/phase/agent-ops-mcp-control-plane/milestones/plane-work-item-webhook-intake.md`
+- Completed task ids:
+  - `loop-idempotency`: PASS; evidence=`agent-task/archive/2026/06/m-plane-work-item-webhook-intake/03+02_idempotency/plan_cloud_G07_1.log`, `agent-task/archive/2026/06/m-plane-work-item-webhook-intake/03+02_idempotency/code_review_cloud_G07_1.log`; verification=`ssh toki@toki-labs.com 'zsh -lc "cd ~/agent-work/nomadcode/services/core && go test -count=1 ./..."'`
+- Not completed task ids: 없음
+
+## 잔여 Nit
+
+- 없음
+
+## 후속 작업
+
+- 없음

agent-task/archive/2026/06/m-plane-work-item-webhook-intake/03+02_idempotency/plan_cloud_G07_1.log

@@ -0,0 +1,210 @@
+<!-- task=m-plane-work-item-webhook-intake/03+02_idempotency plan=1 tag=REVIEW_WEBHOOK_IDEMPOTENCY -->
+
+# Plan - REVIEW_WEBHOOK_IDEMPOTENCY
+
+## 이 파일을 읽는 구현 에이전트에게
+
+구현 완료 전 active `CODE_REVIEW-*-G??.md`의 구현 에이전트 소유 섹션을 실제 구현 내용, 설계 결정, 검증 출력으로 반드시 채운다. 구현 중 사용자만 결정할 수 있는 외부 환경, secret, scope 충돌이 생기면 리뷰 stub의 `사용자 리뷰 요청` 섹션에 증거와 재개 조건을 기록하고 멈춘다. 구현 중 사용자에게 직접 질문하거나 선택지를 제시하거나 `request_user_input`을 호출하지 않는다. 후속 에이전트가 명령 재실행이나 산출물 수집으로 닫을 수 있는 증거 공백은 사용자 리뷰 요청이 아니다. Finalization은 code-review-skill 전용이다.
+
+## 배경
+
+G06 리뷰에서 duplicate external ref task row는 DB upsert로 막히지만, work item pipeline의 existing-task 조회와 workspace provision/slot reservation이 원자적으로 묶이지 않아 동시 duplicate webhook에서 slot side effect가 새어 나갈 수 있음이 확인됐다. 또한 storage/query 멱등성과 handler duplicate 테스트가 계획의 검증 강도에 못 미쳤다. 이 follow-up은 같은 Milestone Task의 멱등성 요구를 닫기 위한 최소 보완이다.
+
+## 사용자 리뷰 요청 흐름
+
+구현 중 blocker는 active review stub의 `사용자 리뷰 요청` 섹션에 `agent-ops/skills/common/_templates/implementation-user-review-request-section.md` 형식으로 기록한다. 직접 사용자 프롬프트는 금지하며, code-review가 USER_REVIEW 작성 여부를 판단한다.
+
+## Roadmap Targets
+
+- Milestone: `agent-roadmap/phase/agent-ops-mcp-control-plane/milestones/plane-work-item-webhook-intake.md`
+- Task ids:
+  - `loop-idempotency`: duplicate webhook, provider retry, NomadCode self actor event를 중복 creation task 없이 처리한다.
+- Completion mode: check-on-pass
+
+## 분석 결과
+
+### 읽은 파일
+
+- `agent-ops/skills/common/code-review/SKILL.md`
+- `agent-ops/skills/common/plan/SKILL.md`
+- `agent-ops/skills/common/_templates/implementation-user-review-request-section.md`
+- `agent-ops/rules/project/domain/core/rules.md`
+- `agent-test/local/rules.md`
+- `agent-test/local/core-smoke.md`
+- `agent-roadmap/current.md`
+- `agent-task/archive/2026/06/m-plane-work-item-webhook-intake/02+01_trigger_dispatch/complete.log`
+- `agent-task/m-plane-work-item-webhook-intake/03+02_idempotency/plan_cloud_G06_0.log`
+- `agent-task/m-plane-work-item-webhook-intake/03+02_idempotency/code_review_cloud_G06_0.log`
+- `services/core/internal/workitempipeline/service.go`
+- `services/core/internal/workitempipeline/service_test.go`
+- `services/core/internal/http/plane_webhook_test.go`
+- `services/core/internal/storage/store.go`
+- `services/core/internal/storage/workspace_slots_test.go`
+- `services/core/internal/workflow/service.go`
+- `services/core/internal/workflow/service_test.go`
+- `services/core/queries/tasks.sql`
+- `services/core/queries/workspace_slots.sql`
+- `services/core/internal/db/tasks.sql.go`
+- `services/core/internal/db/workspace_slots.sql.go`
+- `services/core/migrations/00008_add_task_external_ref_unique.sql`
+
+### 테스트 환경 규칙
+
+- `test_env=local`로 적용한다.
+- `agent-test/local/rules.md`와 `agent-test/local/core-smoke.md`를 읽었다.
+- core 변경의 필수 baseline은 `cd services/core && go test ./...`이고, fresh 실행이 필요하므로 `-count=1`을 사용한다.
+- standard remote runner 기준으로 `ssh toki@toki-labs.com 'zsh -lc "cd ~/agent-work/nomadcode/services/core && go test -count=1 ./..."'`도 최종 검증에 포함한다.
+- sqlc generated code가 바뀌면 `cd services/core && go run github.com/sqlc-dev/sqlc/cmd/sqlc@v1.31.1 generate`를 사용한다.
+- `<확인 필요>` 값은 없다.
+
+### 테스트 커버리지 공백
+
+- `services/core/internal/workitempipeline/service.go:92`의 existing-task 선조회는 sequential duplicate에는 동작하지만 concurrent duplicate에는 같은 external ref 작업을 직렬화하지 않는다.
+- `services/core/internal/workitempipeline/service_test.go:636`과 `:687`은 이미 존재하는 task를 반환하는 경로만 검증하고, 두 요청이 동시에 `pgx.ErrNoRows`를 본 뒤 slot을 두 번 예약하는 실패 모드를 검증하지 않는다.
+- `services/core/queries/tasks.sql:4`의 upsert와 `services/core/internal/storage/store.go:60`의 lookup에 대응하는 task storage/query test가 없다.
+- `services/core/internal/http/plane_webhook_test.go:526`의 fake는 같은 task를 무조건 반환하므로 duplicate webhook이 duplicate side effect를 만들지 않는다는 증거가 아니다.
+
+### 심볼 참조
+
+- renamed/removed symbol 없음.
+- `workitempipeline.TaskCreator` call site는 `services/core/internal/http/handlers.go`에서 `workflow.Service`를 주입하는 path와 `services/core/internal/workitempipeline/service_test.go`의 fake 구현이다.
+- `GetTaskByExternalRef` 참조는 `services/core/internal/workflow/service.go`, `services/core/internal/storage/store.go`, `services/core/internal/workitempipeline/service.go`, 관련 tests에 있다.
+
+### 분할 판단
+
+- shared task group: `m-plane-work-item-webhook-intake`
+- selected subtask: `03+02_idempotency`
+- predecessor `02`는 `agent-task/archive/2026/06/m-plane-work-item-webhook-intake/02+01_trigger_dispatch/complete.log`로 충족됐다.
+- code-review follow-up은 기존 실패 지점만 보완하므로 같은 subtask 디렉터리에서 plan=1로 이어간다.
+
+### 범위 결정 근거
+
+- historical duplicate data backfill이나 운영 DB 수동 정리는 이 follow-up 범위가 아니다.
+- roadmap 문서 갱신은 code-review PASS completion event 이후 런타임 책임이므로 이 plan에서 수정하지 않는다.
+- Plane live smoke는 sibling `04+01,02,03_live_smoke` 범위다. 여기서는 deterministic unit/storage tests와 remote core baseline만 닫는다.
+
+### 빌드 등급
+
+- `cloud-G07`: storage/migration idempotency와 concurrent side-effect prevention이 결합된 correctness failure이며, 기존 증거가 테스트 통과에도 놓친 동시성 edge를 다룬다.
+
+## 의존 관계 및 구현 순서
+
+- `02+01_trigger_dispatch` 완료 근거: `agent-task/archive/2026/06/m-plane-work-item-webhook-intake/02+01_trigger_dispatch/complete.log`.
+- 먼저 provider/id scoped atomic boundary를 구현하고, 그 다음 storage/http/pipeline tests를 보강한다.
+
+## 구현 체크리스트
+
+- [ ] 동일 external provider/id의 work item creation path를 원자화하여 concurrent duplicate webhook에서도 workspace provision과 slot reservation이 한 번만 발생하게 한다. 검증: 같은 provider/work item을 동시에 처리해도 동일 task를 반환하고 `ReserveWorkspaceSlot` 호출이 1회다.
+- [ ] storage/query duplicate external ref test와 meaningful duplicate webhook/pipeline tests를 추가하여 `CreateTask` upsert, `GetTaskByExternalRef`, duplicate webhook side-effect 방지를 검증한다.
+- [ ] sqlc/gofmt/focused tests/local full core/remote full core 검증을 실행하고 실제 stdout/stderr를 review stub에 기록한다.
+- [ ] CODE_REVIEW-*-G??.md의 구현 에이전트 소유 섹션을 실제 구현 내용과 검증 출력으로 채운다. 이 항목이 완료되기 전에는 구현이 완료된 것이 아니다.
+
+## [REVIEW_WEBHOOK_IDEMPOTENCY-1] Atomic External Ref Side-Effect Boundary
+
+### 문제
+
+`services/core/internal/workitempipeline/service.go:92`는 existing task를 먼저 조회하지만, `services/core/internal/workitempipeline/service.go:127`부터 workspace provision과 slot reservation을 수행한 뒤 `services/core/internal/workitempipeline/service.go:155`에서 task create/upsert를 호출한다. 동일 external ref 요청 2개가 동시에 들어오면 둘 다 no-row를 보고 각각 slot을 예약할 수 있다.
+
+### 해결 방법
+
+Provider/id가 있는 work item creation path를 provider/id scoped lock 또는 원자적 claim으로 감싼다. 권장 구현은 `workflow.Service`/`storage.Store` 쪽에 external ref lock helper를 추가하고, `workitempipeline.Service.CreateTaskFromWorkItem`에서 같은 external ref에 대해 아래 흐름 전체를 직렬화하는 것이다.
+
+```go
+// services/core/internal/workitempipeline/service.go:91
+if input.Ref.Provider != "" && input.Ref.ID != "" {
+    return s.tasks.WithExternalRefLock(ctx, string(input.Ref.Provider), input.Ref.ID, func(ctx context.Context) (storage.Task, error) {
+        existing, err := s.tasks.GetTaskByExternalRef(ctx, string(input.Ref.Provider), input.Ref.ID)
+        if err == nil {
+            return existing, nil
+        }
+        if !errors.Is(err, pgx.ErrNoRows) {
+            return storage.Task{}, err
+        }
+        return s.createTaskFromWorkItemUnlocked(ctx, input)
+    })
+}
+return s.createTaskFromWorkItemUnlocked(ctx, input)
+```
+
+Lock 구현은 Postgres advisory lock처럼 process 간에도 같은 provider/id를 직렬화하는 방식을 우선한다. session lock을 쓰면 dedicated connection을 acquire하고 `defer`로 unlock/release를 보장한다. in-process mutex만 쓰는 경우 multi-instance core에서 같은 문제가 남으므로, 명확한 이유와 남은 위험을 review stub에 기록해야 한다.
+
+### 수정 파일 및 체크리스트
+
+- [ ] `services/core/internal/storage/store.go`: provider/id scoped lock 또는 atomic claim helper를 expose한다.
+- [ ] `services/core/internal/workflow/service.go`: pipeline이 사용할 lock/atomic helper를 validation 포함으로 연결한다.
+- [ ] `services/core/internal/workitempipeline/service.go`: external ref가 있을 때 locked section 안에서 second lookup, trigger evaluation, workspace provision, slot reservation, create를 수행한다.
+- [ ] `services/core/internal/workitempipeline/service_test.go`: concurrent duplicate test를 추가한다.
+
+### 테스트 작성
+
+- Required: `TestCreateTaskFromWorkItemConcurrentDuplicateExternalRefReservesOneSlot`.
+- 두 goroutine이 같은 `workitem.Ref{Provider:"plane", ID:"work-1"}`로 진입하도록 만들고, 첫 번째 create 이후 두 번째는 existing task를 반환해야 한다.
+- assertion: returned task IDs are equal, `ReserveWorkspaceSlot` count is exactly 1, `CreateTask` count is exactly 1.
+
+### 중간 검증
+
+```bash
+cd services/core && go test -count=1 ./internal/workitempipeline
+```
+
+Expected: concurrent duplicate regression test를 포함해 package가 통과한다.
+
+## [REVIEW_WEBHOOK_IDEMPOTENCY-2] Storage And Webhook Evidence
+
+### 문제
+
+`services/core/queries/tasks.sql:4`와 `services/core/internal/storage/store.go:60`의 storage/query 멱등성은 production behavior의 핵심인데 task storage test가 없다. `services/core/internal/http/plane_webhook_test.go:526`의 duplicate fake는 같은 task를 무조건 반환하므로 duplicate side effect 방지 증거가 되지 않는다.
+
+### 해결 방법
+
+`services/core/internal/storage/workspace_slots_test.go`의 mock DBTX 패턴을 재사용해 task query shape와 arguments를 검증하는 storage test를 추가한다. 실제 DB harness가 없으면 최소한 generated query가 `ON CONFLICT (external_provider, external_id)`와 partial predicate를 포함하고, `GetTaskByExternalRef`가 provider/id args를 그대로 전달함을 확인한다.
+
+HTTP duplicate test는 무조건 같은 task를 반환하는 fake 대신 stateful fake를 사용한다. 첫 번째 호출은 create side effect를 기록하고, 두 번째 동일 ref 호출은 existing task 반환만 기록하게 해 duplicate create count가 1임을 검증한다. 더 적합하면 handler-level test는 stable dispatch ref 검증으로 좁히고, side effect count는 pipeline concurrent test에서 강하게 검증한다.
+
+### 수정 파일 및 체크리스트
+
+- [ ] `services/core/internal/storage/task_store_test.go`: `CreateTask` upsert query shape, args, default metadata/external metadata, `GetTaskByExternalRef` args를 검증한다.
+- [ ] `services/core/internal/http/plane_webhook_test.go`: duplicate test가 fake의 무조건 동일 반환에 의존하지 않도록 수정한다.
+- [ ] `services/core/internal/workitempipeline/service_test.go`: side effect 방지 assertion을 concurrent case와 연결한다.
+- [ ] `services/core/internal/db/tasks.sql.go`: sqlc regenerate 결과가 `queries/tasks.sql`과 일치하는지 확인한다.
+
+### 테스트 작성
+
+- Required: `TestCreateTask_UsesExternalRefUpsert`.
+- Required: `TestGetTaskByExternalRef_PassesProviderAndID`.
+- Required: duplicate webhook test update. 이름은 유지해도 되지만, assertion은 duplicate create side effect count가 1임을 증명해야 한다.
+
+### 중간 검증
+
+```bash
+cd services/core && go test -count=1 ./internal/storage ./internal/http ./internal/workitempipeline
+```
+
+Expected: focused packages가 통과한다.
+
+## 수정 파일 요약
+
+| 파일 | 항목 |
+|------|------|
+| `services/core/internal/storage/store.go` | REVIEW_WEBHOOK_IDEMPOTENCY-1 |
+| `services/core/internal/workflow/service.go` | REVIEW_WEBHOOK_IDEMPOTENCY-1 |
+| `services/core/internal/workitempipeline/service.go` | REVIEW_WEBHOOK_IDEMPOTENCY-1 |
+| `services/core/internal/workitempipeline/service_test.go` | REVIEW_WEBHOOK_IDEMPOTENCY-1, REVIEW_WEBHOOK_IDEMPOTENCY-2 |
+| `services/core/internal/storage/task_store_test.go` | REVIEW_WEBHOOK_IDEMPOTENCY-2 |
+| `services/core/internal/http/plane_webhook_test.go` | REVIEW_WEBHOOK_IDEMPOTENCY-2 |
+| `services/core/internal/db/tasks.sql.go` | REVIEW_WEBHOOK_IDEMPOTENCY-2 |
+
+## 최종 검증
+
+```bash
+cd services/core && go run github.com/sqlc-dev/sqlc/cmd/sqlc@v1.31.1 generate
+cd services/core && gofmt -w internal/storage/store.go internal/workflow/service.go internal/workitempipeline/service.go internal/workitempipeline/service_test.go internal/storage/task_store_test.go internal/http/plane_webhook_test.go internal/db/tasks.sql.go
+cd services/core && go test -count=1 ./internal/storage ./internal/workflow ./internal/workitempipeline ./internal/http
+cd services/core && go test -count=1 ./...
+tar -C services/core -czf - . | ssh toki@toki-labs.com 'mkdir -p ~/agent-work/nomadcode/services/core && tar -C ~/agent-work/nomadcode/services/core -xzf -'
+ssh toki@toki-labs.com 'zsh -lc "cd ~/agent-work/nomadcode/services/core && go test -count=1 ./..."'
+```
+
+Expected: sqlc generation succeeds, focused tests pass, local full core tests pass, remote full core tests pass.
+
+모든 코드 변경 완료 후 반드시 `CODE_REVIEW-*-G??.md`의 구현 에이전트 소유 섹션을 채운다. 이 파일 작성이 구현의 마지막 단계다.

services/core/internal/http/plane_webhook_test.go

@@ -10,6 +10,7 @@ import (
 	"net/http/httptest"
 	"os"
 	"strings"
+	"sync"
 	"testing"
 
 	"github.com/nomadcode/nomadcode-core/internal/storage"
@@ -517,21 +518,33 @@ func testPlaneSignature(secret, body string) string {
 	return hex.EncodeToString(mac.Sum(nil))
 }
 
-type idempotentFakeWorkItemCreator struct {
+// statefulWorkItemCreator simulates pipeline-level idempotency: the first call
-	task      storage.Task
+// for a given provider/id creates a task and records it; subsequent calls for
-	err       error
+// the same ref return the existing task without incrementing createCount.
-	callCount int
+type statefulWorkItemCreator struct {
+	mu          sync.Mutex
+	created     map[string]storage.Task
+	createCount int
 }
 
-func (f *idempotentFakeWorkItemCreator) CreateTaskFromWorkItem(_ context.Context, input workitempipeline.CreateTaskInput) (storage.Task, error) {
+func (f *statefulWorkItemCreator) CreateTaskFromWorkItem(_ context.Context, input workitempipeline.CreateTaskInput) (storage.Task, error) {
-	f.callCount++
+	f.mu.Lock()
-	return f.task, f.err
+	defer f.mu.Unlock()
+	key := string(input.Ref.Provider) + "\x00" + input.Ref.ID
+	if existing, ok := f.created[key]; ok {
+		return existing, nil
+	}
+	f.createCount++
+	task := storage.Task{ID: "task-1", Status: "created"}
+	if f.created == nil {
+		f.created = make(map[string]storage.Task)
+	}
+	f.created[key] = task
+	return task, nil
 }
 
 func TestReceivePlaneWebhookDuplicateSignedEvent(t *testing.T) {
-	creator := &idempotentFakeWorkItemCreator{
+	creator := &statefulWorkItemCreator{}
-		task: storage.Task{ID: "task-1", Status: "created"},
-	}
 	h := newHandlerForTest(withPlaneCreator(creator))
 	h.SetPlaneWebhookSecret("secret")
 	h.SetPlaneWebhookDispatchConfig(planeDispatchTestConfig())
@@ -549,7 +562,7 @@ func TestReceivePlaneWebhookDuplicateSignedEvent(t *testing.T) {
 		t.Fatalf("expected accepted status with task-1, got %v", resp1)
 	}
 
-	// Second call (duplicate webhook retry)
+	// Second call (duplicate webhook retry) — same task returned, no new creation.
 	req2, rec2 := signedPlaneWebhookRequest(body)
 	h.ReceivePlaneWebhook(rec2, req2)
 	if rec2.Code != http.StatusAccepted {
@@ -560,8 +573,8 @@ func TestReceivePlaneWebhookDuplicateSignedEvent(t *testing.T) {
 		t.Fatalf("expected accepted status with task-1, got %v", resp2)
 	}
 
-	if creator.callCount != 2 {
+	if creator.createCount != 1 {
-		t.Errorf("expected CreateTaskFromWorkItem to be called twice, got %d", creator.callCount)
+		t.Errorf("expected exactly one task creation, got %d (duplicate side effect leaked)", creator.createCount)
 	}
 }
 

services/core/internal/storage/task_store_test.go

@@ -0,0 +1,234 @@
+package storage
+
+import (
+	"context"
+	"encoding/json"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/jackc/pgx/v5"
+	"github.com/jackc/pgx/v5/pgconn"
+	"github.com/nomadcode/nomadcode-core/internal/db"
+)
+
+// ---- mock DBTX for task queries ----
+
+type taskFakeResult struct {
+	row db.Task
+	err error
+}
+
+type taskQueryCall struct {
+	sql  string
+	args []interface{}
+}
+
+type taskMockDBTX struct {
+	result     taskFakeResult
+	queryCalls []taskQueryCall
+}
+
+func (m *taskMockDBTX) Exec(_ context.Context, _ string, _ ...interface{}) (pgconn.CommandTag, error) {
+	return pgconn.NewCommandTag("OK"), nil
+}
+
+func (m *taskMockDBTX) Query(_ context.Context, _ string, _ ...interface{}) (pgx.Rows, error) {
+	return nil, nil
+}
+
+func (m *taskMockDBTX) QueryRow(_ context.Context, sql string, args ...interface{}) pgx.Row {
+	m.queryCalls = append(m.queryCalls, taskQueryCall{sql: sql, args: args})
+	return &taskMockRow{result: m.result}
+}
+
+type taskMockRow struct {
+	result taskFakeResult
+}
+
+func (r *taskMockRow) Scan(dest ...interface{}) error {
+	if r.result.err != nil {
+		return r.result.err
+	}
+	scanTask(r.result.row, dest)
+	return nil
+}
+
+// scanTask fills scan destinations in the column order returned by task queries.
+func scanTask(row db.Task, dest []interface{}) {
+	if len(dest) < 14 {
+		return
+	}
+	if p, ok := dest[0].(*string); ok {
+		*p = row.ID
+	}
+	if p, ok := dest[1].(*string); ok {
+		*p = row.Title
+	}
+	if p, ok := dest[2].(*string); ok {
+		*p = row.Source
+	}
+	if p, ok := dest[3].(*string); ok {
+		*p = row.Status
+	}
+	if p, ok := dest[4].(*json.RawMessage); ok {
+		*p = row.Payload
+	}
+	if p, ok := dest[5].(*json.RawMessage); ok {
+		*p = row.Result
+	}
+	if p, ok := dest[6].(**string); ok {
+		*p = row.Error
+	}
+	if p, ok := dest[7].(*time.Time); ok {
+		*p = row.CreatedAt
+	}
+	if p, ok := dest[8].(*time.Time); ok {
+		*p = row.UpdatedAt
+	}
+	if p, ok := dest[9].(**string); ok {
+		*p = row.ExternalProvider
+	}
+	if p, ok := dest[10].(**string); ok {
+		*p = row.ExternalID
+	}
+	if p, ok := dest[11].(**string); ok {
+		*p = row.ExternalUrl
+	}
+	if p, ok := dest[12].(*json.RawMessage); ok {
+		*p = row.ExternalMetadata
+	}
+	if p, ok := dest[13].(*json.RawMessage); ok {
+		*p = row.Metadata
+	}
+}
+
+func taskSQLContains(calls []taskQueryCall, sub string) bool {
+	for _, c := range calls {
+		if strings.Contains(c.sql, sub) {
+			return true
+		}
+	}
+	return false
+}
+
+func newTaskStore(m *taskMockDBTX) *Store {
+	return NewStoreWithQueries(db.New(m))
+}
+
+func strPtr(s string) *string { return &s }
+
+// ---- CreateTask ----
+
+func TestCreateTask_UsesExternalRefUpsert(t *testing.T) {
+	provider := "plane"
+	id := "work-1"
+	m := &taskMockDBTX{result: taskFakeResult{row: db.Task{
+		ID:               "task-abc",
+		Title:            "Fix issue",
+		Source:           "plane",
+		Status:           "pending",
+		Payload:          json.RawMessage(`{}`),
+		ExternalProvider: strPtr(provider),
+		ExternalID:       strPtr(id),
+		ExternalMetadata: json.RawMessage(`{}`),
+		Metadata:         json.RawMessage(`{}`),
+	}}}
+	store := newTaskStore(m)
+
+	task, err := store.CreateTask(context.Background(), CreateTaskInput{
+		Title:            "Fix issue",
+		Source:           "plane",
+		Payload:          json.RawMessage(`{}`),
+		ExternalProvider: strPtr(provider),
+		ExternalID:       strPtr(id),
+	})
+	if err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
+	if task.ID != "task-abc" {
+		t.Errorf("task ID: got %q, want task-abc", task.ID)
+	}
+
+	if len(m.queryCalls) == 0 {
+		t.Fatal("expected a query call")
+	}
+	sql := m.queryCalls[0].sql
+	for _, fragment := range []string{
+		"ON CONFLICT (external_provider, external_id)",
+		"WHERE external_provider IS NOT NULL AND external_id IS NOT NULL",
+		"DO UPDATE SET updated_at",
+	} {
+		if !strings.Contains(sql, fragment) {
+			t.Errorf("expected CreateTask SQL to contain %q\ngot: %s", fragment, sql)
+		}
+	}
+
+	args := m.queryCalls[0].args
+	if len(args) < 8 {
+		t.Fatalf("expected at least 8 args, got %d", len(args))
+	}
+	// args order: title, source, payload, metadata, external_provider, external_id, external_url, external_metadata
+	if got, ok := args[4].(*string); !ok || *got != provider {
+		t.Errorf("arg[4] external_provider: got %v, want %q", args[4], provider)
+	}
+	if got, ok := args[5].(*string); !ok || *got != id {
+		t.Errorf("arg[5] external_id: got %v, want %q", args[5], id)
+	}
+}
+
+// ---- GetTaskByExternalRef ----
+
+func TestGetTaskByExternalRef_PassesProviderAndID(t *testing.T) {
+	provider := "plane"
+	id := "work-1"
+	m := &taskMockDBTX{result: taskFakeResult{row: db.Task{
+		ID:               "task-xyz",
+		Title:            "Existing task",
+		Source:           "plane",
+		Status:           "pending",
+		Payload:          json.RawMessage(`{}`),
+		ExternalProvider: strPtr(provider),
+		ExternalID:       strPtr(id),
+		ExternalMetadata: json.RawMessage(`{}`),
+		Metadata:         json.RawMessage(`{}`),
+	}}}
+	store := newTaskStore(m)
+
+	task, err := store.GetTaskByExternalRef(context.Background(), provider, id)
+	if err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
+	if task.ID != "task-xyz" {
+		t.Errorf("task ID: got %q, want task-xyz", task.ID)
+	}
+
+	if len(m.queryCalls) == 0 {
+		t.Fatal("expected a query call")
+	}
+	sql := m.queryCalls[0].sql
+	if !strings.Contains(sql, "WHERE external_provider = $1 AND external_id = $2") {
+		t.Errorf("expected WHERE clause with provider and id params\ngot: %s", sql)
+	}
+
+	args := m.queryCalls[0].args
+	if len(args) < 2 {
+		t.Fatalf("expected 2 args, got %d", len(args))
+	}
+	if got, ok := args[0].(*string); !ok || *got != provider {
+		t.Errorf("arg[0] external_provider: got %v, want %q", args[0], provider)
+	}
+	if got, ok := args[1].(*string); !ok || *got != id {
+		t.Errorf("arg[1] external_id: got %v, want %q", args[1], id)
+	}
+}
+
+func TestGetTaskByExternalRef_PropagatesNoRows(t *testing.T) {
+	m := &taskMockDBTX{result: taskFakeResult{err: pgx.ErrNoRows}}
+	store := newTaskStore(m)
+
+	_, err := store.GetTaskByExternalRef(context.Background(), "plane", "missing")
+	if err != pgx.ErrNoRows {
+		t.Fatalf("expected pgx.ErrNoRows, got %v", err)
+	}
+}

services/core/internal/workflow/service.go

@@ -6,10 +6,51 @@ import (
 	"errors"
 	"log/slog"
 	"strings"
+	"sync"
 
 	"github.com/nomadcode/nomadcode-core/internal/storage"
 )
 
+// externalRefLocks provides per-key mutual exclusion for external ref operations.
+// It is safe for concurrent use. In-process only: multi-instance deployments
+// still require a distributed lock (e.g. Postgres advisory lock) to prevent
+// cross-instance duplicate side effects.
+type externalRefLocks struct {
+	mu      sync.Mutex
+	entries map[string]*refEntry
+}
+
+type refEntry struct {
+	mu      sync.Mutex
+	holders int
+}
+
+func (r *externalRefLocks) acquire(key string) *refEntry {
+	r.mu.Lock()
+	if r.entries == nil {
+		r.entries = make(map[string]*refEntry)
+	}
+	e, ok := r.entries[key]
+	if !ok {
+		e = &refEntry{}
+		r.entries[key] = e
+	}
+	e.holders++
+	r.mu.Unlock()
+	e.mu.Lock()
+	return e
+}
+
+func (r *externalRefLocks) release(key string, e *refEntry) {
+	e.mu.Unlock()
+	r.mu.Lock()
+	e.holders--
+	if e.holders == 0 {
+		delete(r.entries, key)
+	}
+	r.mu.Unlock()
+}
+
 var (
 	ErrInvalidTaskInput      = errors.New("invalid task input")
 	ErrTaskCannotBeEnqueued  = errors.New("task cannot be enqueued in current status")
@@ -21,10 +62,11 @@ type TaskEnqueuer interface {
 }
 
 type Service struct {
-	store     *storage.Store
+	store       *storage.Store
-	enqueuer  TaskEnqueuer
+	enqueuer    TaskEnqueuer
-	logger    *slog.Logger
+	logger      *slog.Logger
-	lifecycle *Lifecycle
+	lifecycle   *Lifecycle
+	extRefLocks externalRefLocks
 }
 
 func NewService(store *storage.Store, enqueuer TaskEnqueuer, logger *slog.Logger) *Service {
@@ -36,6 +78,13 @@ func NewService(store *storage.Store, enqueuer TaskEnqueuer, logger *slog.Logger
 	}
 }
 
+func (s *Service) WithExternalRefLock(ctx context.Context, provider, id string, fn func(ctx context.Context) (storage.Task, error)) (storage.Task, error) {
+	key := provider + "\x00" + id
+	e := s.extRefLocks.acquire(key)
+	defer s.extRefLocks.release(key, e)
+	return fn(ctx)
+}
+
 func (s *Service) CreateTask(ctx context.Context, input CreateTaskInput) (storage.Task, error) {
 	title := strings.TrimSpace(input.Title)
 	source := strings.TrimSpace(input.Source)

services/core/internal/workitempipeline/service.go

@@ -38,6 +38,7 @@ type CreateTaskInput struct {
 type TaskCreator interface {
 	CreateTask(ctx context.Context, input workflow.CreateTaskInput) (storage.Task, error)
 	GetTaskByExternalRef(ctx context.Context, provider, id string) (storage.Task, error)
+	WithExternalRefLock(ctx context.Context, provider, id string, fn func(ctx context.Context) (storage.Task, error)) (storage.Task, error)
 }
 
 // ProjectBinding is the resolved project sync binding for a work item: the
@@ -89,18 +90,22 @@ func New(reader workitem.Reader, binder ProjectBinder, tasks TaskCreator) *Servi
 }
 
 func (s *Service) CreateTaskFromWorkItem(ctx context.Context, input CreateTaskInput) (storage.Task, error) {
-	// Idempotency check: see if a task already exists for this external reference.
 	if input.Ref.Provider != "" && input.Ref.ID != "" {
-		existing, err := s.tasks.GetTaskByExternalRef(ctx, string(input.Ref.Provider), input.Ref.ID)
+		return s.tasks.WithExternalRefLock(ctx, string(input.Ref.Provider), input.Ref.ID, func(ctx context.Context) (storage.Task, error) {
-		if err == nil {
+			existing, err := s.tasks.GetTaskByExternalRef(ctx, string(input.Ref.Provider), input.Ref.ID)
-			return existing, nil
+			if err == nil {
-		}
+				return existing, nil
-		if !errors.Is(err, pgx.ErrNoRows) {
+			}
-			return storage.Task{}, err
+			if !errors.Is(err, pgx.ErrNoRows) {
-		}
+				return storage.Task{}, err
+			}
+			return s.createTaskFromWorkItemUnlocked(ctx, input)
+		})
 	}
+	return s.createTaskFromWorkItemUnlocked(ctx, input)
+}
 
-	// Resolve the project sync binding first.
+func (s *Service) createTaskFromWorkItemUnlocked(ctx context.Context, input CreateTaskInput) (storage.Task, error) {
 	if s.binder == nil {
 		return storage.Task{}, ErrProjectSyncNotConfigured
 	}

services/core/internal/workitempipeline/service_test.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"encoding/json"
 	"errors"
+	"sync"
 	"testing"
 
 	"github.com/jackc/pgx/v5"
@@ -50,6 +51,10 @@ func (f *fakeTaskCreator) GetTaskByExternalRef(_ context.Context, provider, id s
 	return f.getTaskTask, f.getTaskErr
 }
 
+func (f *fakeTaskCreator) WithExternalRefLock(_ context.Context, _, _ string, fn func(context.Context) (storage.Task, error)) (storage.Task, error) {
+	return fn(context.Background())
+}
+
 // fakeBinder resolves a single fixed binding and reserves a single fixed slot,
 // recording what it was called with so tests can assert ordering and inputs.
 type fakeBinder struct {
@@ -726,3 +731,139 @@ func TestCreateTaskFromWorkItemDuplicateExternalRefDoesNotReserveSlot(t *testing
 func testOptionalString(s string) *string {
 	return &s
 }
+
+// ---- concurrent duplicate test helpers ----
+
+// concurrentFakeTaskCreator implements TaskCreator with real per-key locking and
+// stateful GetTaskByExternalRef so concurrent duplicate requests see the task
+// created by whichever goroutine won the lock first.
+type concurrentFakeTaskCreator struct {
+	stateMu     sync.Mutex
+	tasks       map[string]storage.Task
+	createCount int
+
+	lockMu   sync.Mutex
+	lockKeys map[string]*sync.Mutex
+	lockRefs map[string]int
+}
+
+func (f *concurrentFakeTaskCreator) acquireKey(key string) *sync.Mutex {
+	f.lockMu.Lock()
+	if f.lockKeys == nil {
+		f.lockKeys = make(map[string]*sync.Mutex)
+		f.lockRefs = make(map[string]int)
+	}
+	mu, ok := f.lockKeys[key]
+	if !ok {
+		mu = &sync.Mutex{}
+		f.lockKeys[key] = mu
+	}
+	f.lockRefs[key]++
+	f.lockMu.Unlock()
+	mu.Lock()
+	return mu
+}
+
+func (f *concurrentFakeTaskCreator) releaseKey(key string, mu *sync.Mutex) {
+	mu.Unlock()
+	f.lockMu.Lock()
+	f.lockRefs[key]--
+	if f.lockRefs[key] == 0 {
+		delete(f.lockKeys, key)
+		delete(f.lockRefs, key)
+	}
+	f.lockMu.Unlock()
+}
+
+func (f *concurrentFakeTaskCreator) WithExternalRefLock(_ context.Context, provider, id string, fn func(context.Context) (storage.Task, error)) (storage.Task, error) {
+	key := provider + "\x00" + id
+	mu := f.acquireKey(key)
+	defer f.releaseKey(key, mu)
+	return fn(context.Background())
+}
+
+func (f *concurrentFakeTaskCreator) GetTaskByExternalRef(_ context.Context, provider, id string) (storage.Task, error) {
+	f.stateMu.Lock()
+	defer f.stateMu.Unlock()
+	if f.tasks == nil {
+		return storage.Task{}, pgx.ErrNoRows
+	}
+	key := provider + "\x00" + id
+	if t, ok := f.tasks[key]; ok {
+		return t, nil
+	}
+	return storage.Task{}, pgx.ErrNoRows
+}
+
+func (f *concurrentFakeTaskCreator) CreateTask(_ context.Context, input workflow.CreateTaskInput) (storage.Task, error) {
+	f.stateMu.Lock()
+	defer f.stateMu.Unlock()
+	f.createCount++
+	task := storage.Task{ID: "task-concurrent-1", Status: "pending"}
+	if input.External != nil && input.External.Provider != "" && input.External.ID != "" {
+		if f.tasks == nil {
+			f.tasks = make(map[string]storage.Task)
+		}
+		f.tasks[input.External.Provider+"\x00"+input.External.ID] = task
+	}
+	return task, nil
+}
+
+// countingFakeBinder is a thread-safe binder that counts slot reservations.
+type countingFakeBinder struct {
+	binding      ProjectBinding
+	slot         projectsync.WorkspaceSlot
+	mu           sync.Mutex
+	reserveCount int
+}
+
+func (f *countingFakeBinder) ResolveProjectBinding(_ context.Context, _ workitem.Ref) (ProjectBinding, error) {
+	return f.binding, nil
+}
+
+func (f *countingFakeBinder) EnsureProjectWorkspace(_ context.Context, _ ProjectBinding) error {
+	return nil
+}
+
+func (f *countingFakeBinder) ReserveWorkspaceSlot(_ context.Context, _ int64) (projectsync.WorkspaceSlot, error) {
+	f.mu.Lock()
+	f.reserveCount++
+	f.mu.Unlock()
+	return f.slot, nil
+}
+
+func TestCreateTaskFromWorkItemConcurrentDuplicateExternalRefReservesOneSlot(t *testing.T) {
+	ref := workitem.Ref{Provider: "plane", Tenant: "acme", Project: "proj-1", ID: "work-1"}
+	reader := &fakeReader{item: workitem.WorkItem{Ref: ref, Title: "Concurrent Task"}}
+	creator := &concurrentFakeTaskCreator{}
+	binder := &countingFakeBinder{binding: validBinding(), slot: reservedSlot()}
+	svc := New(reader, binder, creator)
+
+	var wg sync.WaitGroup
+	results := make([]storage.Task, 2)
+	errs := make([]error, 2)
+
+	wg.Add(2)
+	for i := range 2 {
+		go func(i int) {
+			defer wg.Done()
+			results[i], errs[i] = svc.CreateTaskFromWorkItem(context.Background(), CreateTaskInput{Ref: ref})
+		}(i)
+	}
+	wg.Wait()
+
+	for i, err := range errs {
+		if err != nil {
+			t.Errorf("goroutine %d: unexpected error: %v", i, err)
+		}
+	}
+	if results[0].ID != results[1].ID {
+		t.Errorf("expected same task ID from both goroutines, got %q and %q", results[0].ID, results[1].ID)
+	}
+	if creator.createCount != 1 {
+		t.Errorf("CreateTask call count: got %d, want 1", creator.createCount)
+	}
+	if binder.reserveCount != 1 {
+		t.Errorf("ReserveWorkspaceSlot call count: got %d, want 1", binder.reserveCount)
+	}
+}