전 과정 생성 이미지 DB 영속화 — 구현 계획서

Task A1 — 스키마 마이그레이션 + 모델 필드 + graph service 정합

Files: Create migration, Modify app/models/project.py, (정합 확인) app/services/pipeline_graph_service.py.

• Produces: ImageAsset.input_image_ids(Text JSON), stage(Text), pipeline_role(Text), is_intermediate(Boolean default false), generation_call_id(Text), candidate_index(Integer), pipeline_metadata_json(Text). 인덱스 3개.

1. ☐ 모델 필드 추가(project.py ImageAsset): 위 7개 Column(..., nullable=True)(is_intermediate는 default=False, nullable=False). reference_image_ids 주석/의미 불변.
2. ☐ 마이그 작성: alembic revision -m "image_asset generation cols" → op.add_column 7개 + op.create_index (project_id,episode_id,is_intermediate)/generation_call_id/(project_id,episode_id,pipeline_role). idempotent: inspector = sa.inspect(conn); existing = {c['name'] for c in inspector.get_columns('image_asset')} 가드. downgrade=drop.
3. ☐ 적용: cd backend && .venv/bin/alembic upgrade head → 에러 없음. psql \d image_asset로 컬럼/인덱스 확인.
4. ☐ graph service 정합: 신규 컬럼은 캔버스 후속(Phase C)에서 사용. 지금은 모델 추가가 기존 _asset_to_node/쿼리에 영향 없음만 확인 → pytest tests/services/test_pipeline_graph_service.py -q 21 green 유지.
5. ☐ startup 검증: uvicorn 재기동 → _migrations fail-fast 통과(schema drift 0).
6. ☐ 커밋: feat(image-capture): image_asset 생성 영속화 컬럼 + 마이그.

Task A2 — GenerationContext (contextvar + scope + worker bind) TDD

Files: Create image_capture/context.py, tests/services/image_capture/test_context.py.

1. ☐ 실패 테스트: current_context() 기본 None. with generation_context("p","e","stage") as q: 안에서 current_context().stage=="stage", q is current_context().queue. 블록 밖 다시 None(reset). bind_context(ctx)로 다른 스레드 함수에서 동일 ctx 보임(thread target에 명시 전파).
2. ☐ 실행→실패 확인: pytest tests/services/image_capture/test_context.py -q → ImportError.
3. ☐ 구현: ContextVar[GenerationContext|None], generation_context contextmanager(set token → try/yield queue → finally queue.flush() + reset token), bind_context(set/reset helper). flush() 실패도 non-fatal(로그만).
4. ☐ 통과: pytest green.
5. ☐ 커밋.

Task A3 — file-backed spool TDD

Files: Create image_capture/spool.py, tests/.../test_spool.py.

1. ☐ 실패 테스트: p = write_spool(b"PNGDATA", project_id, stage) → 파일 존재 + 바이트 일치 + 경로가 spool 디렉터리 하위. 두 번 호출 시 경로 충돌 없음(uuid).
2. ☐ 실행→실패.
3. ☐ 구현: projects_root/.capture_spool/<project>/<uuid>.png에 Path.write_bytes. 디렉터리 mkdir(parents, exist_ok). 반환=절대경로.
4. ☐ 통과 → 커밋.

Task A4 — CaptureQueue.flush (DB insert + rename, 독립 세션, diagnostics) TDD

Files: Create image_capture/queue.py, tests/.../test_queue.py(PG theroad_test).

1. ☐ 실패 테스트: 시드 project/episode → q=CaptureQueue(ctx); spool 파일 1개 append(meta: role, input_image_ids, ...) → q.flush()==1; DB에 ImageAsset(is_intermediate=True, pipeline_role=role, input_image_ids=json, file_path=final) 1행 + spool→final rename됨 + spool 파일 사라짐. flush 두 번째 호출=0(큐 비움). DB 에러 시 flush가 raise 안 함(독립 세션, non-fatal) + diag 증가.
2. ☐ 실행→실패.
3. ☐ 구현: append=리스트 적재. flush=독립 SessionLocal() 열고 각 항목 ImageAsset insert + spool→episodes/<eid>/images/generated/<stage>/<uuid>.png rename, commit, close. 전체 try/except(실패=rollback+diag, raise 안 함). 큐 클리어.
4. ☐ 통과 → 커밋.

Task A5 — sink (capture_generated_image / capture_artifact) TDD

Files: Create image_capture/sink.py, tests/.../test_sink.py.

1. ☐ 실패 테스트: (1) capture scope 밖에서 capture_generated_image(b"x", role="r") 호출 → spool/큐 변화 0 + CAPTURE_DIAG["skipped_no_context"]++(default-capture-off=중복 0 보장). (2) with generation_context(...) as q: 안에서 호출 → spool 1 + q에 append 1(meta role/input_image_ids 보존). (3) capture_artifact도 동일 경로. (4) ★dedup 테스트: scope가 열려도 호출 자체가 explicit capture라야 저장(최종물 경로는 scope 안 열거나 호출 안 함 → 0행).
2. ☐ 실행→실패.
3. ☐ 구현: ctx=current_context(); ctx None이면 diag++ & return(저장 안 함). 있으면 write_spool→ctx.queue.append(spool, meta). 전체 try/except non-fatal.
4. ☐ 통과 → 커밋.

Task A6 — 통합 결정론 테스트 + Codex 리뷰 (Phase A 종료)

1. ☐ E2E(단위): with generation_context("p","e","pose_guide") as q: capture_generated_image(png, role="pose_guide", input_image_ids=[bg,base]) → 블록 종료 finally flush → DB에 intermediate 1행, input_image_ids 보존, 파일 존재.
2. ☐ ★중복 0 회귀: 기존 최종물 경로(scope 미사용) 시뮬 → image_asset 중복 0.
3. ☐ 전체: pytest tests/services/image_capture -q + 기존 회귀(graph service 21, image steps) green.
4. ☐ Codex 리뷰(답신 %0) → 반영 → 사용자 GO 커밋.

Task B0 — gpt primitive wrapper call_gpt_image_bytes 결정론 TDD

Files: Create app/modules/llm/gpt_image_primitive.py(또는 image_capture/gpt_primitive.py), tests/.../test_gpt_image_primitive.py.

1. ☐ 실패 테스트(mock OpenAI client): call_gpt_image_bytes(client, mode, prompt, ref_paths=[2개], call_kwargs={size,quality,'n':1}, capture_role) → client.images.edit 1회 호출. ★file handle 객체 동일성 비교는 어려우므로 method name + prompt/model/size/quality/n + image 인자 개수 + 파일명/bytes 순서를 assert(Codex). ref_paths=[1개]→image=단일핸들(리스트 아님). ref_paths=[]→images.generate(image 인자 없음). 반환=decode된 bytes. scope 열림 시 capture enqueue 1, 닫힘 시 0. len(png)<min_capture_bytes(1024)→capture skip+diag 이지만 bytes 는 그대로 반환(사이트 semantics 불변).
2. ☐ 구현: ExitStack 으로 ref_paths open(len≥2=list / ==1=단일핸들 / ==0=generate, 각 사이트 현 형태 정확 재현), resp=client.images.edit|generate(model=model, image?=..., prompt=prompt, **call_kwargs), b64=resp.data[0].b64_json, img=base64.b64decode(b64), min-size 통과 시 capture_generated_image(img, role=capture_role, prompt=prompt, generation_call_id=None, pipeline_metadata={budget_source,...}), return img(+필요시 b64/raw resp). budget reserve·out_path write·검증·retry 는 호출 안 함(사이트 유지). ImageCallBudgetExceeded 등은 wrapper가 삼키지 않고 전파.
3. ☐ 통과 → (커밋은 B1 사이트 치환과 함께 또는 별도, 사용자 GO).

Task B1 — gpt 5개 사이트 치환 mock call-args TDD + 실이미지 canary

Files: Modify background_render.py·background_chain_render.py·floor_plan_render.py·location_floor_plan.py·space_set_bg_provider.py. 사이트별 test_*_callargs.py.

1. ☐ 치환: 각 사이트의 openai_client.images.{edit,generate}(...) → b64 decode 블록을 png=call_gpt_image_bytes(..., call_kwargs=<그 사이트가 현재 넘기는 정확한 dict>, budget_source=<분기 라벨>, capture_role=<표 role>) 로 교체. reserve_current_call(분기별 라벨)·out_path.write_bytes·empty/too-small 검증·info dict 반환·retry/sanitizer 루프는 그대로. space_set 는 call_kwargs 에 quality·n 미포함(현 호출 보존), decode→png_bytes 후 SpaceSetBgProviderError empty 의미 보존.
2. ☐ call-args TDD: 각 사이트를 mock client 로 호출 → images.edit/generate 가 치환 전과 동일 kwargs로 호출됨 assert(특히 space_set 에 quality 미주입 / edit_single 이 단일 핸들 / edit_multi 가 리스트). budget reserve 라벨 문자열 불변 assert.
3. ☐ 실이미지 canary(scratchpad 드라이버, 실 gpt-image-2 소수 호출, scope OPEN 1개 + scope 닫힘 1개): scope 열고 producer 1회 → image_asset intermediate 1행(role/stage 정확)+산출 PNG 육안 타당. scope 닫고 1회 → image_asset 행 0(production 경로 무변). 갤러리 8898. ★커밋금지 드라이버.
4. ☐ 회귀: 기존 background/floor_plan/space_set step 결정론 테스트 green(동작 무변). Codex 검토 → 사용자 GO 커밋.

Task B2 — Gemini generate_image 계측 (★수렴점) mock TDD + 중복0

Files: Modify gemini_image_client.py(:277-292), test_gemini_image_capture.py.

1. ☐ 구현: :292 직전(success if b64: 블록, 기존 log_llm_call/_tracer.log 뒤)에서 call_id = log_llm_call(...) 반환 회수(현 폐기됨, signature 확인), img = base64.b64decode(b64) 1회 바인딩, capture_generated_image(img, role=<neutral/scope>, input_image_ids=ref_image_ids 라벨, prompt=prompt, generation_call_id=call_id, pipeline_metadata=self._ctx 부분집합), return img, elapsed_ms. reserve/budget·log·tracer 인자·순서 불변.
2. ☐ 중복0 테스트(★핵심): scope 닫힘에서 generate_image 호출(최종 reference/scene 경로 시뮬) → image_asset 행 0(default-off). scope 열림에서만 1행.
3. ☐ mock TDD: capture 가 마지막 statement·non-raise, byte 동일(decode 1회 재사용), error/refusal/no-image 경로는 capture 미도달.
4. ☐ canary: scope 열어 reference 1장 생성 → intermediate 1행+육안. Codex 검토 → 커밋 GO.

Task B3 — gemini_i2i_editor 직접 REST 계측 mock TDD

1. ☐ 구현: _gemini_generate_content :196 img = base64.b64decode(b64); capture_generated_image(img, role='i2i_edit', prompt=prompt, input_image_ids=None, pipeline_metadata={model,aspect_ratio}); return img. 단일점이 edit_angle/color/combined 4경로 모두 커버 → 호출측 중복계측 불필요. retry/safety 가드·return 흐름 불변.
2. ☐ camera diagram = 포함 Codex 합의: _create_camera_diagram :94-96 png=buf.getvalue() 직후 capture_artifact(png, role='angle_camera_diagram', pipeline_metadata={horizontal_deg,vertical_deg,zoom}); return png. underlay/marking 과 같은 "모델 입력" 클래스지만 "전부" 원칙상 포함, UI 필터로 접음.
3. ☐ mock TDD(call-args/decode 1회/non-raise) + canary(scope 열어 i2i variation 1장).

Task B4 — fal apply_fal_angle 계측 mock TDD

1. ☐ 구현: fal_angle_helpers.py :233 data = dl.read() 직후 capture_generated_image(data, role='fal_angle', prompt=None, pipeline_metadata={h,v,z}), return 흐름 불변. budget reserve 라벨 fal_angle_helpers.fal_run 불변.
2. ☐ mock TDD(다운로드 바이트 동일·non-raise) + canary(scope 열어 fal 앵글 1장, FAL_AI_ENABLED=true 필요).

Task B5 — 비모델 PIL capture_artifact 결정론 TDD

1. ☐ registered underlay: build_registered_pose_guide :285 make_registration_underlay 직후 capture_artifact(underlay_png, role='registered_pose_underlay', pipeline_metadata={group_id,bg_key,variant})(except 밖).
2. ☐ zoom crop: generate_continuity_crop_png :331 crop_bytes 직후 capture_artifact(crop_bytes, role='zoom_continuity_crop', pipeline_metadata={group_id,source,scene_index,shot_index,bbox}).
3. ☐ space_set marking: 디스크 only → caller space_set_bg_step.py :335 core.mark_fp(...) 직후 capture_artifact(Path(marked).read_bytes(), role='space_set_marked_fp', pipeline_metadata={group_id,room,num}). mark_fp 내부 무수정.
4. ☐ dwelling zone 주석(신규): _annotate_fp :300 fp.save(out_path) 직후 capture_artifact(Path(out_path).read_bytes(), role='dwelling_zone_annotated_fp', pipeline_metadata={verification_only:true,...})(또는 BytesIO 경유).
5. ☐ birdseye legacy(선택, 미호출): :611 return 직전 1줄 — 최하 우선순위.
6. ☐ 결정론 TDD: scope 열림 시 각 capture_artifact 가 spool+queue append 1, 닫힘 시 0. PIL save 인자/순서 불변 assert(read_bytes=동일 파일 재독).

Task B6 — 통합 검증 + 제외 가드 + Codex 리뷰 (Phase B 종료)

1. ☐ ★static/audit allowlist 가드 Codex 추가: rg/AST 기반 테스트로 openai_client.images.{generate,edit} 직접 호출이 wrapper 밖에 0 임을 잠금 — background_render/background_chain_render/floor_plan_render/location_floor_plan/space_set_bg_provider 전체 치환 보증(미래 누락·재도입 차단).
2. ☐ 제외 negative 가드: png_metadata.embed_png_metadata / thumbnail·export / consumer write_bytes 사이트에 capture 호출 0(이중캡처 0 증명).
3. ☐ no-scope E2E: 모든 wrapper/Gemini/FAL/artifact 를 호출해도 CAPTURE_DIAG['skipped_no_context']↑ + image_asset 신규 intermediate 0(production 바이트·행 무변).
4. ☐ open-scope E2E: 각 producer 별 최소 1개 → intermediate row 1개씩 생기고, consumer write_bytes 로 추가 row 0(producer 수렴점에서만 잡힘 증명).
5. ☐ 전체: pytest tests/services/image_capture + 신규 callargs + 기존 image step 회귀 green.
6. ☐ producer별 실이미지 canary(scope 열어 작은 synthetic, 8898 갤러리 육안) + Codex 리뷰(답신 %0) → 반영 → 사용자 GO 커밋(producer 묶음별 scope).

Task C0 — 토대: queue scene_index fold + capture threading primitive 결정론 TDD

1. ☐ queue scene_index fold: flush 에서 ctx.scene_index is not None 이면 pipeline_metadata(없으면 {})에 "scene_index" 병합 후 pipeline_metadata_json 에 저장. 테스트: scope scene_index=12 + capture → row 의 pipeline_metadata_json 에 scene_index:12. scene_index None 이면 키 부재(기존 동작 보존).
2. ☐ capture_input_image_ids threading: call_gpt_image_bytes(..., capture_input_image_ids: list[str]|None=None) → capture_generated_image(input_image_ids=capture_input_image_ids, ...). generate_floor_plan_image 도 동일 optional 인자 받아 thread. mock call-args TDD: default None 일 때 images.{edit,generate} kwargs + capture 인자가 치환 전과 동일(byte-identical), 명시 전달 시 input_image_ids 가 capture 까지 도달. 기존 5사이트 회귀 green.
3. ☐ 커밋(사용자 GO): feat(image-capture): Phase C 토대 — scene_index fold + input_image_ids threading.

Task C1 — zoom continuity scope (cleanest, end-to-end proof) 결정론 TDD + canary

1. ☐ source_asset_id 해석: 단건=_load_source_primary_bytes(:200) 를 (bytes, asset_id) 반환으로 확장(이미 asset row 조회 중 — asset.id 추가만). batch=메인스레드 _persist_primary_and_track(selected_id) 에서 still_id→primary_asset_id 맵 구축(scene_paths_by_index_by_id 평행) → worker 주입(worker DB 금지 준수).
2. ☐ 함수 내부 crop scope(dedup 경계): generate_continuity_crop_png 에 still_id/source_asset_id param 추가. crop 생성+capture 섹션(:320-340)을 with generation_context(project_id, episode_id, stage='zoom_continuity', still_id, scene_index, shot_index) 로 감싸고 fill(:342+)은 with 밖. crop 의 capture_artifact 에 input_image_ids=[source_asset_id](None 이면 생략) + pipeline_metadata={bbox,group_id,source}.
3. ☐ 결정론 TDD: (a) scope 안 crop capture → intermediate row 1 (input_image_ids=[source]). (b) ★fill 의 gemini 호출은 scope 밖이라 image_asset 추가 0(중복 0, 최종물 무변) — gemini client mock 으로 ctx None 확인. (c) source_asset_id None fallback 비차단.
4. ☐ canary(scratchpad 드라이버, 작은 zoom 재생성): crop intermediate 1행(role/stage/input_image_ids 정확)+캔버스에서 source→crop 상류 엣지 육안, 최종 scene asset 행수/바이트 무변. 8898 갤러리. ★커밋금지 드라이버.

Task C2 — registered pose guide scope 결정론 TDD + canary

1. ☐ scope 배선: attach_registered_pose_guide_ref 호출(single coordinator:832 / batch :1426)을 with generation_context(project_id, episode_id, stage='registered_pose_guide', still_id=still_data.get('id'), scene_index, shot_index) 로 감쌈. ★batch 는 worker 본문 내라 그 thread 에서 직접 열림(bind_context 불필요).
2. ☐ role 교정: build_registered_pose_guide 가 generate_floor_plan_image 에 capture_role='registered_pose_guide' 전달(현재 미전달 → default floor_plan_image 오라벨). underlay(make_registration_underlay:106 capture_artifact)는 scope 안에서 자동 영속화.
3. ☐ input_image_ids v1 defer: bg plate UUID 는 worker DB 금지 + bg_map 무 UUID → v1 은 pipeline_metadata={bg_id, group_id, variant} 만(UUID 는 Phase D 메인스레드 resolve). 부분 lineage 허용.
4. ☐ 결정론 TDD: flag ON 시 guide=role 'registered_pose_guide' + underlay 2자산 capture. flag OFF 시 attach no-op → capture 0(byte-identical). cache_hit 시 generate 미호출 → 중복 capture 0(멱등).
5. ☐ canary(flag ON, 작은 immobilized 그룹): guide+underlay intermediate 행 + 캔버스 노드 육안. 8898.

Task C3 — outdoor 3-stage scope 결정론 TDD + canary

1. ☐ scope 배선: _generate_shared_model_guides(메인 스레드) 에서 base_fn(:686)=stage='outdoor_site_layout', scene_index/shot_index=None(group 공통), guide_fn(:720)=scene_index=g_si, shot_index=g_shi. still_id=None(scene_still 이전 단계).
2. ☐ stage별 role: provider 헬퍼(_t2i_image/_edit_image_with_ref) + generate_floor_plan_image 에 capture_role 전달 → aerial_base / shot_blocking / camera_sketch 구분(default floor_plan_image 오염 교정).
3. ☐ chaining v1: A→B→C UUID lineage 는 불가 → pipeline_metadata={group_id, location_id, stage} 로 사후 그룹 엣지(Decision #2 ⓑ). scene_index 는 C0 fold 로 metadata.
4. ☐ 결정론 TDD: flag ON 시 stage별 role 정확. flag OFF 시 scope 미발동(byte-identical). base_cache hit 시 base capture 1회만.
5. ☐ canary(flag ON, 야외 cross-shot 그룹 실재 필요): aerial/blocking/sketch intermediate 행 + 캔버스 group 육안. 8898.

Task C4 — E2E canary + dedup 가드 + Codex 리뷰 (Phase C 종료)

1. ☐ 작은 에피소드 재생성(3 사이트 flag ON): DB intermediate 행 생성 확인(role/stage/input_image_ids/scene_index metadata 정확).
2. ☐ ★중복 0 가드: 최종 scene/reference/fp asset 행수·바이트 무변(특히 zoom fill 이 intermediate 로 안 샘). is_intermediate=True 만 신규.
3. ☐ 캔버스 lineage 육안(8898 외부 웹서버, open 금지): source→crop 엣지, pose/outdoor 중간 노드 group 표시.
4. ☐ 전체 회귀(image_capture + zoom/pose/outdoor step) green + Codex 리뷰(답신 %0) → 반영 → 사용자 GO 커밋(사이트 묶음별 scope).