Merge remote-tracking branch 'origin/main' into feature/site-features-and-mockup

.env.example

@@ -83,9 +83,17 @@ ALIMTALK_TEMPLATE_CODE=
 # 묶어야 한다. 빌더에서 코드를 받아 채널에 한 번 입력하는 절차다.
 # ★ 이 값이 비면 연결 화면이 아예 안 뜬다 — 어디에 코드를 칠지 말해 줄 수 없는데
 #   코드만 발급하면 사장님에게는 고장난 화면이다.
-# ★ 사장님 대화창(에이전트). 기본 꺼짐 — 카카오톡 채널이 준비되기 전에는 띄우지 않는다.
+# 사장님 대화창(에이전트). 1=사용, 0=감춤.
-#   코드는 다 있지만 채널 없이 열어 두면 사장님에게는 어디에도 닿지 않는 입구다.
+# ★ 1 이어도 LLM 키가 없으면 안 열린다 — 키 없는 환경에서 켜 둔 채 잊어도
-AGENT_CHAT_ENABLED=0
+#   "눌러도 안 되는 입구" 가 생기지 않는다.
+AGENT_CHAT_ENABLED=1
+# 카카오톡 채널 웹훅(오픈빌더 스킬 서버). ★ 오픈빌더는 서명을 주지 않는다 —
+# URL 만 알면 누구나 때릴 수 있고 발화자 id 를 위조하면 그 사장님 행세를 한다.
+# 비우면 웹훅 엔드포인트가 404 다(반쯤 열린 상태를 만들지 않는다).
+#   만드는 법: python -c "import secrets; print(secrets.token_urlsafe(32))"
+KAKAO_WEBHOOK_SECRET=
+# 우리 봇이 맞는지 한 겹 더. 오발송을 거르는 용도라 비워도 된다.
+KAKAO_BOT_ID=
 KAKAO_CHANNEL_PUBLIC_ID=
 KAKAO_LINK_CODE_TTL_MIN=10
 KAKAO_LINK_MAX_ATTEMPTS=5

AGENTS.md

@@ -169,9 +169,19 @@
   `services/*` 를 통과한다. `collect_service.store_facts` 가 크롤러에 걸어 둔 그 문이다.
 - **카카오 채널 발화자는 우리 `user_id` 가 아니다** — 채널 단위 익명 키다.
   `owner_kakao_links` 매핑 없이 발화자를 믿으면 **채널 진입점만 소유자 범위 밖**에 놓인다.
-- **에이전트 화면은 2026-09-21 기준 감춰져 있다** — `AGENT_CHAT_ENABLED=0`(기본) ·
+- **★ 카카오 웹훅은 서명이 없다 — 시크릿이 유일한 문이다.** 오픈빌더는 서명을 주지 않아서,
-  `KAKAO_CHANNEL_PUBLIC_ID` 빔. 카카오톡 채널이 보류돼서이고 **코드는 멀쩡하다**.
+  URL 만 알면 누구나 때릴 수 있고 `userRequest.user.id` 를 위조하면 **그 사장님 행세**를 한다.
-  "기능이 없다" 고 판단해 지우지 않는다([AGENT.md](docs/AGENT.md)).
+  `KAKAO_WEBHOOK_SECRET` 이 비면 엔드포인트가 **404**(401 은 존재를 알린다).
+- **확인 대기에 만료가 없으면 묵은 발행이 돈다** — 카카오톡은 앞선 답을 되돌려 주지 않아
+  서버가 pending 을 들고 있는다. `pending_expires_at`(3분)을 빼면 한참 뒤의 "네" 한 마디에
+  실행된다([AGENT.md](docs/AGENT.md)).
+- **바로가기 라벨과 '예' 로 읽는 말이 어긋나면 눌러도 안 먹는다** — 사장님은 버튼이 고장난
+  줄 안다. `channel.py` 의 `CONFIRM_LABEL` 상수를 쓰고 문자열을 손으로 적지 않는다.
+- **에이전트 대화창은 스위치와 LLM 키를 둘 다 본다**(`AGENT_CHAT_ENABLED`, 기본 `1`).
+  키만 보면 "잠시 닫아 두기" 가 키를 지우는 일이 되어 소개문·사진분류까지 꺼지고,
+  스위치만 보면 키 없는 환경에 **눌러도 안 되는 입구**가 생긴다.
+  카카오 연결 카드는 `KAKAO_CHANNEL_PUBLIC_ID` 가 비면 감춰진다 —
+  웹훅(4단계)이 없어 코드를 보내도 연결이 완성되지 않기 때문이다([AGENT.md](docs/AGENT.md)).
 - **에이전트 등급을 모델이 정하게 두지 않는다** — 확인이 필요한 행위인지는 `services/agent/tools.py`
   레지스트리가 못 박는다. 응답 스키마에 그 칸을 만들면 프롬프트에 끼어든 한 줄이 확인 절차를 건너뛴다.
 - **실행 결과 문구를 LLM 이 쓰게 두지 않는다** — 모델은 **하지 않은 일을 했다고 말할 수 있고**,

docs/AGENT.md

@@ -4,25 +4,27 @@
 **에이전트는 카카오톡 안에 있지 않다.** 카톡은 입구 하나이고, 같은 에이전트가 빌더 화면에도
 붙는다. 그래야 채널·챗봇 심사 전에 전부 검증된다.
 
-지금까지 만든 것은 **1단계(신원 연결)** 와 **2단계(도구·런타임·빌더 채팅창)** 다.
+1단계(신원 연결) · 2단계(도구·런타임·빌더 채팅창) · **4단계(카카오 웹훅)** 을 만들었다.
-카카오 채널 웹훅은 아직 없다.
+남은 것은 오픈빌더 챗봇 등록(우리가 못 하는 일)과 도구 늘리기다.
 
-## ★ 지금은 화면에서 감춰져 있다 (2026-09-21 보류)
+## 화면 스위치
 
-카카오톡 채널 개설이 **법인폰 본인인증**에 걸려 보류됐다. 채널이 없으면 이 기능은
+| 화면 | 여는 조건 | 지금 |
-사장님에게 **어디에도 닿지 않는 입구**다 — 열어 두면 "되는 기능" 으로 오해한다.
+|---|---|---|
+| 대화창(`AgentChatDock`) | `AGENT_CHAT_ENABLED=1`(기본) **그리고** LLM 키 | 열림 |
+| 연결 카드(`KakaoChannelCard`) | `KAKAO_CHANNEL_PUBLIC_ID` 가 채워짐 | 채널 ID 미설정 |
 
-| 화면 | 감추는 조건 |
+★ 스위치와 키를 **둘 다** 본다(`runtime.is_configured`). 키만 보면 "잠시 닫아 두기" 를 키를
-|---|---|
+지워서 해야 하고 그러면 소개문·사진분류까지 같이 꺼진다. 스위치만 보면 키 없는 환경에서
-| 대화창(`AgentChatDock`) | `AGENT_CHAT_ENABLED=0` (기본값) |
+**눌러도 안 되는 입구**가 생긴다.
-| 연결 카드(`KakaoChannelCard`) | `KAKAO_CHANNEL_PUBLIC_ID` 가 빔 (기본값) |
 
-**코드는 그대로 두고 설정으로만 닫았다.** 채널이 준비되면 값 둘을 채우고 다시 띄우면 된다 —
+★ 2026-09-21 에 카카오 채널 개설이 법인폰 본인인증에 걸려 한 번 닫았고,
-되돌릴 때 커밋을 되짚지 않는다. 서버도 함께 닫힌다(`runtime.is_configured()` 가 스위치를
+인증이 끝나 2026-09-22 에 다시 열었다. **그때도 코드는 지우지 않고 값만 바꿨다** —
-보므로, 화면을 우회해 API 를 직접 불러도 `AGENT_NOT_CONFIGURED` 다).
+닫고 여는 일이 커밋을 되짚는 일이 되면 안 된다.
 
-★ Threads 카드는 반대로 '자리는 두고 버튼만 죽이는' 쪽이다. 저쪽은 사장님이 **곧 쓸 수 있는**
+★ 연결 카드를 '감추는' 쪽으로 둔 것은 Threads 카드('자리는 두고 버튼만 죽인다')와 반대
-기능이라 존재를 알려야 했고, 이쪽은 언제 열릴지 말해 줄 수 없다. 판단이 갈린 이유가 그것이다.
+판단인데 의도한 차이다. 저쪽은 사장님이 곧 쓸 수 있는 기능이라 존재를 알려야 했고,
+이쪽은 웹훅(4단계)이 없어 아직 연결이 **완성되지 않는다**.
 
 ## 왜 신원 연결이 먼저인가
 
@@ -77,13 +79,10 @@ RETURNING user_id;
 **실패는 전부 같은 에러다**(`KAKAO_LINK_CODE_INVALID`). "없는 코드"·"만료"·"시도 초과" 를
 구분해 답하면 6자리 코드의 유효성을 외부에서 탐색할 수 있다.
 
-## ★ 소비 엔드포인트는 아직 없다
+## 코드는 웹훅에서만 소비된다
 
-코드를 소비하는 쪽은 **채널 웹훅**이고, 그 웹훅은 자체 서명 검증을 갖춘 뒤에야 열 수 있다.
+`redeem()` 은 **공개 라우터에 붙어 있지 않다.** 시크릿 검증을 통과한 웹훅 안에서만 불린다 —
-검증 없는 공개 소비 경로를 먼저 만들면 누구나 코드를 대입해 남의 계정에 자기 카톡을 붙인다 —
+검증 없는 공개 소비 경로가 있으면 누구나 6자리를 대입해 남의 계정에 자기 카톡을 붙인다.
-이 표가 막으려던 바로 그 일이다.
-
-지금 `redeem()` 은 서비스 함수로만 있고 라우터에 붙어 있지 않다.
 
 ## API
 
@@ -179,3 +178,96 @@ services/fact_service.py · site_service.py      ★ 게이트가 사는 곳
 검증·출처 필수·정정본 보호가 **아무 증상 없이** 사라진다.
 `collect_service.store_facts` 가 크롤러에 걸어 둔 문과 같은 문이고,
 `tests/test_agent_runtime.py` 가 소스에서 그 호출이 없는지 실제로 검사한다.
+
+
+---
+
+# 4단계 — 카카오 채널 웹훅
+
+```
+router/v1/agent/kakao_bot.py   시크릿 검증 · 카카오 형식 ↔ 우리 모양   ← 카카오를 아는 유일한 파일
+services/agent/channel.py      신원 · 가게 고르기 · 확인 이어받기      ← 카카오를 모른다
+services/agent/runtime.py      그대로 — 채널을 모른다
+```
+
+## ★★ 인증 — 오픈빌더는 서명을 주지 않는다
+
+URL 만 알면 누구나 이 엔드포인트를 때릴 수 있고, `userRequest.user.id` 를 아무 값이나 넣으면
+**그 사장님 행세를 한다.** 신원 연결이 통째로 무의미해지는 자리다.
+
+| 겹 | 방법 |
+|---|---|
+| 1 | 공유 시크릿 — 헤더 `X-Agent-Secret` (`hmac.compare_digest`) |
+| 2 | `KAKAO_BOT_ID` 대조 (시크릿이 아니라 오발송을 거르는 용도, 비워도 됨) |
+| 3 | 헤더를 못 넣을 때만 경로 시크릿 `/webhook/{secret}` — **최후 수단**, 경로는 로그에 남는다 |
+
+★ `KAKAO_WEBHOOK_SECRET` 이 비면 **엔드포인트가 404 다.** 401 로 답하면 "여기 뭔가 있다" 를
+알려 준다. 반쯤 열린 상태를 만들지 않는 것은 Threads 연결과 같은 규칙이다.
+
+## 빌더 화면과 다른 것 셋
+
+| | 빌더 화면 | 카카오톡 |
+|---|---|---|
+| 신원 | 로그인 토큰 | 연결된 발화자 키 → `user_id` (★ **토큰을 발급하지 않는다**) |
+| 대상 | `place_id` 가 URL 에 | 대화에서 고르고 `current_place_id` 에 기억 |
+| 확인 | 프론트가 `{confirm}` 을 되돌려 줌 | **서버가 무엇을 물었는지 들고 있는다** |
+
+★ **연결되자마자 홈페이지 목록을 보여준다.** 연결만 알리고 끝내면 사장님은 어느 홈페이지를
+다루는 대화인지 모른 채 말을 걸게 된다. 목록에는 **발행 여부**를 같이 적는다 — 안 그러면
+고친 것이 손님에게 보이는 줄 안다.
+
+★ 가게가 여럿이면 **바로가기 버튼으로 고르게 한다.** 이름을 외워 치게 하지 않는다.
+임의로 첫 가게를 고르지도 않는다 — 사장님은 엉뚱한 가게를 고쳐 놓고도 모른다.
+"목록"·"가게 바꿔줘" 같은 말로 **언제든 돌아와 바꿀 수 있고**, 이 경로는 LLM 을 부르지 않는다
+(대화가 막혔을 때 처음 찾는 길이라 늘 통해야 하고, 목록 보기에 돈을 쓸 이유도 없다).
+
+★ `pending_expires_at`(3분)이 없으면 **한참 뒤의 "네" 한 마디에 묵은 발행이 실행된다.**
+다른 말을 하면 그 말이 우선이고, 묵은 확인은 그 자리에서 치운다.
+
+★ **바로가기 라벨과 '예' 로 읽는 말이 같아야 한다**(`CONFIRM_LABEL` 등 상수). 어긋나면
+눌러도 안 먹고, 사장님은 버튼이 고장난 줄 안다.
+
+## 5초 벽 — 콜백으로 넘는다
+
+오픈빌더의 스킬 타임아웃은 **5초**다. 넘기면 카카오가 끊어 **말없이 실패하는 봇**이 된다.
+
+★ **실측(2026-09-22): 실제 프롬프트는 4초를 넘겼다.** 개발 중 잰 1.3~2.4초는 항목 두 개짜리
+장난감 프롬프트였고, 진짜는 업종 필드 43개 + fact 수십 개가 실린다. 작은 표본으로 잰 수치를
+상한 근거로 삼으면 이렇게 틀린다.
+
+→ 오픈빌더 스킬 설정에서 **콜백 사용**을 켜면 요청에 `userRequest.callbackUrl` 이 실려 온다.
+
+```
+카카오 → 우리   발화 + callbackUrl
+우리 → 카카오   {"version":"2.0","useCallback":true,"data":{"text":"확인하고 있어요…"}}   (즉답)
+              … 백그라운드에서 답을 만든다 (상한 45초)
+우리 → 카카오   POST callbackUrl  {"version":"2.0","template":{…}}                      (완성분)
+```
+
+★ 콜백 주소는 **1분 · 1회**만 유효하다. 전송에 실패해도 **재시도하지 않는다** — 두 번째 POST 는
+어차피 거절되고, 사장님에게는 이미 "확인하고 있어요" 가 가 있다.
+
+★ 콜백이 꺼져 있으면 예전처럼 동기로 답하되 `DEADLINE_SEC = 4.5` 로 끊는다.
+무거운 잡(BUILD)은 큐에 넣고 즉답하는 구조라 어느 쪽에서도 걸리지 않는다.
+
+★ 어떤 실패도 **HTTP 200 + 안내 문구**로 답한다. 메신저에서는 500 도 침묵으로 보인다.
+
+## 설정
+
+```
+KAKAO_WEBHOOK_SECRET=     # 비면 웹훅이 404. python -c "import secrets; print(secrets.token_urlsafe(32))"
+KAKAO_BOT_ID=             # 선택
+KAKAO_CHANNEL_PUBLIC_ID=  # 채워야 연결 카드가 뜬다(채널 검색용 아이디, `_` 로 시작)
+```
+
+## 오픈빌더에 등록할 주소
+
+```
+https://<발행호스트>/v1/agent/kakao/webhook
+```
+
+★ 스킬 설정에서 **커스텀 헤더**를 넣을 수 있으면 `X-Agent-Secret` 을 쓰고, 못 넣으면
+`/v1/agent/kakao/webhook/<시크릿>` 을 쓴다.
+
+★ **채널 '채팅' 과 '챗봇(오픈빌더 스킬)' 은 다른 기능이다.** 채팅만 켜면 발화가 우리에게
+오지 않는다 — 웹훅이 붙는 쪽은 챗봇이고, 봇을 만들어 채널에 연결해야 한다.

docs/DEVLOG.md

@@ -1,5 +1,87 @@
 # 개발 일지
 
+## 2026-09-22 — 카톡 5초 벽을 콜백으로 넘는다
+
+실제 카톡에서 "시설 편의에서 바비큐 이용 문구 빼줘" 가 **"확인하는 데 시간이 조금 걸리네요"**
+로 끝났다. 타임아웃이었다.
+
+★ **작은 표본으로 잰 수치를 상한 근거로 삼은 것이 틀렸다.** 개발 중 잰 1.3~2.4초는 업종 필드
+두 개짜리 장난감 프롬프트였고, 진짜 요청에는 필드 43개 + fact 수십 개가 실린다.
+"여유가 있다" 고 적어 둔 판단이 실사용 첫날에 깨졌다.
+
+**고친 방법** — 오픈빌더 콜백(스킬 타임아웃 5초, 콜백 주소 1분·1회):
+`userRequest.callbackUrl` 이 실려 오면 `{"useCallback": true}` 로 **즉답**하고, 백그라운드에서
+답을 만든 뒤 그 주소로 따로 POST 한다. 콜백이 꺼져 있으면 예전처럼 동기(4.5초 상한).
+
+★ 콜백 전송 실패는 **재시도하지 않는다** — 1회용 주소라 두 번째 POST 는 거절되고, 사장님에게는
+이미 "확인하고 있어요" 가 가 있다.
+
+★ 오픈빌더 스킬 설정에서 **콜백 사용을 켜야** 이 경로가 열린다. 안 켜면 코드가 있어도
+`callbackUrl` 이 안 와서 동기 경로로만 돈다 — 조용히 예전처럼 동작한다.
+
+**검증** — `test_kakao_webhook.py` 24 passed(콜백 3건 추가: 즉답 형식·콜백 전송·전송 실패).
+
+## 2026-09-22 — 카톡 대화에 홈페이지 목록·가게 고르기
+
+실제로 붙여 보니 빠진 것이 드러났다(사장님 지적): 연결은 됐는데 **어느 홈페이지를 다루는
+대화인지 화면이 말해 주지 않았다.** 가게가 하나면 말없이 자동 선택돼 더 모호했다.
+
+- 연결 직후 목록을 보여준다. 하나면 그 이름과 발행 여부를, 여럿이면 **바로가기 버튼**으로 고르게.
+- 목록 줄에 **발행 여부**를 적는다 — 안 그러면 고친 것이 손님에게 보이는 줄 안다.
+- "목록"·"가게 바꿔줘" 등으로 **언제든 돌아와 바꾼다.** ★ 이 경로는 LLM 을 부르지 않는다 —
+  대화가 막혔을 때 처음 찾는 길이라 늘 통해야 하고, 목록 보기에 돈을 쓸 이유가 없다.
+- 목록은 `list_my_sites` 를 쓴다(사업장 목록이 아니라). `/sites` 화면이 같은 이유로 그걸 쓴다 —
+  사장님이 알아야 하는 건 "가게가 있다" 가 아니라 "발행돼 있나" 다.
+
+**검증** — `test_kakao_webhook.py` 21 passed(목록·전환 4건 추가).
+전체 `845 passed / 53 failed`, 53 은 이번 변경 전과 같다.
+
+## 2026-09-22 — 카카오 채널 웹훅(4단계)
+
+카카오톡 채널이 준비돼 웹훅을 만들었다. **런타임은 한 줄도 안 바뀌었다** — 채널을 모르게
+만들어 둔 것이 여기서 값을 했다. 새로 생긴 것은 형식 변환(`kakao_bot.py`)과 대화 상태
+(`channel.py`)뿐이다.
+
+**★★ 인증 — 오픈빌더는 서명을 주지 않는다**
+URL 만 알면 누구나 때릴 수 있고 `userRequest.user.id` 를 위조하면 **그 사장님 행세**를 한다.
+1단계에서 만든 신원 연결이 통째로 무의미해지는 자리다. 공유 시크릿(헤더 `X-Agent-Secret`,
+`hmac.compare_digest`) + 선택적 `KAKAO_BOT_ID` 대조로 막고, 시크릿이 없으면 **엔드포인트가
+404** 다 — 401 은 "여기 뭔가 있다" 를 알려 준다.
+
+**빌더 화면과 다른 것 셋** — 나머지는 `runtime.chat()` 그대로다.
+1. 로그인 토큰이 없다 → 발화자 키로 사장님을 찾는다. ★ **토큰을 발급하지 않는다**
+   (카톡 경로에서 JWT 가 나오면 그게 곧 권한 탈취 경로다)
+2. `place_id` 가 URL 에 없다 → 대화에서 고르고 `current_place_id` 에 기억.
+   ★ 여럿인데 안 정해졌으면 **추측하지 않고 되묻는다**
+3. 확인을 되돌려 줄 프론트가 없다 → 서버가 pending 을 들고 있는다(0022).
+   ★ `pending_expires_at`(3분)이 없으면 **한참 뒤의 "네" 한 마디에 묵은 발행이 돈다**
+
+**5초 벽** — `DEADLINE_SEC=4.0`. 넘기면 카카오가 끊어 말없이 실패하는 봇이 되므로 안내로
+끊는다. 도구 선택 실측이 1.3~2.4초라 여유가 있다. 콜백은 오픈빌더 지원 여부 확인 뒤에.
+어떤 실패도 **200 + 안내 문구**다 — 메신저에서는 500 도 침묵으로 보인다.
+
+**밟은 것** — `DB_SESSION_MNG.execute_lambda` 는 **람다 반환값을 그대로** 준다(CRUD 관례가
+`(ErrorType, 값)`). 우리 람다가 객체만 돌려주자 언패킹에서 TypeError 가 났고, 라우터가 모든
+예외를 삼키는 구조라 화면에는 "지금은 처리할 수 없어요" 한 줄만 보였다 — 원인이 안 보이는 종류다.
+
+**검증** — `test_kakao_webhook.py` 17 passed(시크릿·위조·만료·가게 되묻기·5초·형식 누출).
+전체 `841 passed / 53 failed` 이고 그 53 은 이번 변경 전과 같다.
+
+## 2026-09-22 — 에이전트 대화창 다시 염(기본 켜짐)
+
+카카오톡 채널의 통신사 인증이 끝나 보류를 푼다(사장님 지시). `AGENT_CHAT_ENABLED` 기본값을
+`0` → `1` 로 돌렸다. **코드는 어제도 오늘도 그대로다** — 닫고 여는 일이 커밋을 되짚는 일이
+되면 안 된다는 어제 판단이 하루 만에 값을 쳤다.
+
+★ 기본을 켜도 **LLM 키가 없으면 안 열린다**(`runtime.is_configured` 가 스위치와 키를 둘 다
+본다). 키 없는 환경에서 켜 둔 채 잊어도 "눌러도 안 되는 입구" 가 생기지 않는다.
+
+★ 카카오 연결 카드는 아직 감춰져 있다 — `KAKAO_CHANNEL_PUBLIC_ID` 미설정.
+채우면 코드는 발급되지만 **소비할 웹훅(4단계)이 없어 연결이 완성되지 않는다.**
+채널 '채팅' 과 '챗봇(오픈빌더 스킬)' 은 다른 기능이고, 웹훅이 붙는 쪽은 후자다.
+
+**검증** — `test_agent_runtime`(스위치 테스트를 새 기본값에 맞춰 갱신)·`test_kakao_link` 34 passed.
+
 ## 2026-09-21 — 에이전트 화면 보류: 설정으로 닫는다(코드는 그대로)
 
 카카오톡 채널 개설이 **법인폰 본인인증**에 걸려 보류됐다(사장님 지시: "이 작업은 여기서 딱

docs/MINI_BLOG.md

@@ -106,7 +106,12 @@
 - 본문: 문구 전문 + 승인 링크 + 수정 링크(`blog_jobs._mail_body`)
 - **승인 링크**: `GET /v1/site/post/approve?t=<토큰>` — 로그인 없음, 토큰이 신원. **누르는
   즉시 승인된다**(확인 화면 없음, 2026-09-17 사장님 지시). 토큰은 32바이트 랜덤 → DB 엔
-  sha256 만, **단회용 · 그날 자정(KST) 만료**(`blog_service.issue_token`)
+  sha256 만, **단회용 · 그날 자정(KST) 만료**(`blog_service.issue_token`). 승인 확인
+  화면은 그 업장의 발행된 사이트(미니 블로그 자리, `#blog`)로 5초 뒤 자동 이동한다
+  (2026-09-22 사장님 지시 — `router/v1/site/post.py _page`, `PostService._blog_url`).
+  재발행(BUILD 잡)은 몇 분 걸리므로 5초 뒤에 이 글이 이미 반영돼 있다는 보장은 없다 —
+  그래도 "어디로 가면 보이는지"는 바로 알려준다. 발행된 사이트가 없으면 자동 이동 없이
+  확인 문구만 보여준다
 - **수정 링크**: `{origin}/blog?placeId=&postId=&auto=<그날짜리 JWT>` — 로그인 흐름이다.
   `CreateDayPassToken`(`router/v1/validator/dependencies.py`)이 자정까지만 사는 접근
   토큰을 찍고, 빌더 앱이 그 토큰으로 로그인해 그 글의 편집 모달을 바로 연다

postgres-init/init-data/init.sql

@@ -636,6 +636,11 @@ CREATE TABLE IF NOT EXISTS public.owner_kakao_links (
     status varchar(16) NOT NULL DEFAULT 'PENDING' CHECK (status IN ('PENDING','LINKED','REVOKED')),
     linked_at timestamptz,
     last_seen_at timestamptz,
+    -- 대화 상태(migrations/0022) — 카카오톡은 앞선 답을 되돌려 주지 않는다.
+    current_place_id uuid,
+    pending_tool varchar(40),
+    pending_args jsonb,
+    pending_expires_at timestamptz,
     created_at timestamptz NOT NULL DEFAULT now(),
     updated_at timestamptz NOT NULL DEFAULT now(),
     deleted boolean NOT NULL DEFAULT false

postgres-init/migrations/0022_owner_kakao_links_conversation.sql

@@ -0,0 +1,14 @@
+-- 0022 · owner_kakao_links 에 대화 상태 — 카카오톡은 앞선 답을 되돌려 주지 않는다.
+--
+-- ★ 빌더 화면은 확인(SEMI) 한 바퀴를 프론트가 이어 줬다. `{confirm:{tool,args}}` 를 그대로
+--   돌려보내므로 서버가 아무것도 기억하지 않아도 됐다.
+--   카카오톡에서 돌아오는 것은 **텍스트 한 줄**뿐이다("네, 해주세요"). 그래서 무엇을 물었는지
+--   서버가 들고 있어야 한다.
+--
+-- ★ pending_expires_at 이 없으면 조용히 틀린다: 사장님이 한참 뒤 다른 맥락에서 "네" 라고
+--   치는 순간 **묵은 발행이 실행된다.** 그 사이에 값이 더 바뀌었을 수도 있다.
+ALTER TABLE public.owner_kakao_links
+    ADD COLUMN IF NOT EXISTS current_place_id uuid,
+    ADD COLUMN IF NOT EXISTS pending_tool varchar(40),
+    ADD COLUMN IF NOT EXISTS pending_args jsonb,
+    ADD COLUMN IF NOT EXISTS pending_expires_at timestamptz;

solution/backend/common/database/model/models.py

@@ -721,6 +721,12 @@ class owner_kakao_links(MainTableMixin, MAIN_BASE):
     status = Column(String(16), nullable=False, server_default=text("'PENDING'"))
     linked_at = Column(DateTime(timezone=True), nullable=True)
     last_seen_at = Column(DateTime(timezone=True), nullable=True)
+    # 대화 상태 — 카카오톡은 앞선 답을 되돌려 주지 않는다(빌더 화면은 프론트가 이어 줬다).
+    # ★ pending_expires_at 이 없으면 한참 뒤의 "네" 한 마디에 묵은 발행이 실행된다.
+    current_place_id = Column(UUID(as_uuid=True), nullable=True)
+    pending_tool = Column(String(40), nullable=True)
+    pending_args = Column(JSONB, nullable=True)
+    pending_expires_at = Column(DateTime(timezone=True), nullable=True)
     __table_args__ = (
         Index("uq_kakao_link_user", "user_id", unique=True, postgresql_where=text("deleted=false AND status IN ('PENDING','LINKED')")),
         Index("uq_kakao_link_channel_key", "channel_user_key", unique=True, postgresql_where=text("deleted=false AND status='LINKED'")),

solution/backend/config/agent_config.py

@@ -23,10 +23,22 @@ class AgentConfig(BaseSettings):
     # 코드가 짧아서(사람이 손으로 친다) 무차별 대입이 가능하다. 시도 수로 끊는다.
     KAKAO_LINK_MAX_ATTEMPTS: int = 5
 
-    # ★ 기본 꺼짐. 카카오톡 채널 개설이 법인폰 본인인증에 걸려 보류됐고(2026-09-21),
+    # 빌더 화면의 대화창. 2026-09-21 에 한 번 닫았다가(카카오 채널 보류) 채널 인증이
-    #   채널 없이 대화창만 띄우면 사장님에게는 **어디에도 닿지 않는 입구**가 된다.
+    # 끝나 다시 열었다(2026-09-22).
-    #   코드는 그대로 두고 이 값으로만 연다 — 되돌릴 때 커밋을 되짚지 않아도 된다.
+    # ★ 이 값이 "1" 이어도 **LLM 키가 없으면 안 열린다**(runtime.is_configured 가 둘 다 본다) —
-    AGENT_CHAT_ENABLED: str = "0"
+    #   키 없는 환경에서 켜 둔 채 잊어도 "눌러도 안 되는 입구" 가 생기지 않는다.
+    #   다시 닫을 일이 생기면 이 값만 "0" 으로 되돌린다. 코드를 되짚지 않는다.
+    AGENT_CHAT_ENABLED: str = "1"
+
+    # ★ 카카오 웹훅 인증. **오픈빌더는 서명을 주지 않는다** — URL 만 알면 누구나 이 엔드포인트를
+    #   때릴 수 있고, user.id 를 아무 값이나 넣으면 **그 사장님 행세를 한다.** 신원 연결
+    #   (owner_kakao_links)이 통째로 무의미해진다.
+    #   그래서 이 값이 없으면 **엔드포인트 자체를 띄우지 않는다**(404). 반쯤 열린 상태를
+    #   만들지 않는 것은 Threads 연결과 같은 규칙이다.
+    #   만드는 법: python -c "import secrets; print(secrets.token_urlsafe(32))"
+    KAKAO_WEBHOOK_SECRET: str = ""
+    # 우리 봇이 맞는지 한 겹 더 본다. 시크릿이 아니라 오발송을 거르는 용도라 비워도 된다.
+    KAKAO_BOT_ID: str = ""
 
 
 def get(name, default=""):
@@ -37,6 +49,10 @@ def chat_enabled() -> bool:
     return get("AGENT_CHAT_ENABLED", "0") == "1"
 
 
+def webhook_secret() -> str:
+    return get("KAKAO_WEBHOOK_SECRET")
+
+
 def kakao_link_enabled() -> bool:
     return bool(get("KAKAO_CHANNEL_PUBLIC_ID"))
 

solution/backend/router/router.py

@@ -29,6 +29,7 @@ import router.v1.social.social
 import router.v1.social.oauth
 import router.v1.agent.kakao
 import router.v1.agent.chat
+import router.v1.agent.kakao_bot
 
 API_SERVER_START_TIME = GTime.UTCStr()
 
@@ -142,3 +143,4 @@ app.include_router(router.v1.social.social.router)
 app.include_router(router.v1.social.oauth.router)
 app.include_router(router.v1.agent.kakao.router)
 app.include_router(router.v1.agent.chat.router)
+app.include_router(router.v1.agent.kakao_bot.router)

solution/backend/router/v1/agent/kakao_bot.py

@@ -0,0 +1,147 @@
+"""카카오톡 채널 웹훅(오픈빌더 스킬 서버) — 카카오 형식은 **이 파일 밖으로 나가지 않는다**.
+
+`version: "2.0"` · `simpleText` · `quickReplies` 같은 모양이 서비스 계층에 새면, 다른 채널을
+붙일 때 그걸 전부 걷어내야 한다. 알림톡 어댑터에 건 것과 같은 규칙이다.
+
+★★ **오픈빌더는 서명을 주지 않는다.** URL 만 알면 누구나 이 엔드포인트를 때릴 수 있고,
+   `userRequest.user.id` 를 아무 값이나 넣으면 **그 사장님 행세를 한다** — 신원 연결
+   (`owner_kakao_links`)이 통째로 무의미해진다. 그래서 공유 시크릿을 우리가 직접 댄다.
+   시크릿이 없으면 **엔드포인트 자체를 띄우지 않는다(404)** — 반쯤 열린 상태를 만들지 않는 것은
+   Threads 연결과 같은 규칙이다.
+
+★ 5초 벽: 오픈빌더의 스킬 타임아웃은 **5초**다. 넘기면 카카오가 끊어 사장님에게는
+  **말없이 실패하는 봇**이 된다.
+  → 오픈빌더 스킬 설정에서 **콜백 사용**을 켜면 요청에 `userRequest.callbackUrl` 이 실려 온다.
+    그때는 `{"useCallback": true}` 로 **즉답**하고, 답을 다 만든 뒤 그 주소로 따로 보낸다.
+    콜백 주소는 **1분 · 1회**만 유효하다.
+  → 콜백이 꺼져 있으면 예전처럼 동기로 답하되 `DEADLINE_SEC` 로 끊는다. 실측(2026-09-22):
+    필드 43개 + fact 수십 개가 실린 실제 프롬프트는 4초를 넘겼다 — 개발 중 재본
+    1.3~2.4초는 항목 두 개짜리 장난감 프롬프트였다.
+"""
+
+import asyncio
+import hmac
+
+import httpx
+from fastapi import APIRouter, BackgroundTasks, Header, HTTPException, Request
+
+from common.logger import LOG
+from config import agent_config as config
+from services.agent import channel
+
+router = APIRouter(prefix="/v1/agent/kakao", tags=["Agent"])
+
+# 콜백이 꺼져 있을 때만 쓰는 상한. 카카오가 5초에 끊으므로 그보다 살짝 앞에서 우리가 끊는다 —
+# 침묵보다 "잠시 뒤 다시" 가 낫다.
+DEADLINE_SEC = 4.5
+# 콜백이 켜져 있을 때의 상한. 콜백 주소가 1분간 유효하므로 그 안에서 넉넉히 잡는다.
+CALLBACK_DEADLINE_SEC = 45.0
+
+_TIMEOUT_TEXT = "확인하는 데 시간이 조금 걸리네요. 잠시 뒤 다시 말씀해 주세요."
+_ERROR_TEXT = "지금은 처리할 수 없어요. 잠시 뒤 다시 말씀해 주세요."
+_WAIT_TEXT = "확인하고 있어요. 잠시만 기다려 주세요."
+
+
+def _reply(text: str, quick_replies=None) -> dict:
+    """오픈빌더 스킬 응답(SkillResponse). ★ 카카오 형식을 아는 유일한 함수다."""
+    payload: dict = {"outputs": [{"simpleText": {"text": text}}]}
+    if quick_replies:
+        # 바로가기는 최대 10개. 누르면 그 라벨이 **다음 발화로 그대로 들어온다** —
+        # channel.py 의 _YES/_NO 가 같은 문자열을 알고 있어야 먹는다.
+        payload["quickReplies"] = [
+            {"label": label, "action": "message", "messageText": label} for label in quick_replies[:10]
+        ]
+    return {"version": "2.0", "template": payload}
+
+
+def _authorize(secret_in_path: str | None, header_secret: str | None, body: dict) -> None:
+    expected = config.webhook_secret()
+    if not expected:
+        # 설정이 없으면 이 기능은 존재하지 않는다. 401 로 답하면 엔드포인트의 존재를 알린다.
+        raise HTTPException(404)
+    given = header_secret or secret_in_path or ""
+    if not hmac.compare_digest(given, expected):
+        LOG.w("[agent/kakao] 웹훅 시크릿 불일치 — 거절")
+        raise HTTPException(404)
+
+    # 한 겹 더. 시크릿이 아니라 오발송을 거르는 용도라 비워 두면 검사하지 않는다.
+    bot_id = config.get("KAKAO_BOT_ID")
+    if bot_id and (body.get("bot") or {}).get("id") != bot_id:
+        LOG.w("[agent/kakao] 다른 봇의 요청 — 거절")
+        raise HTTPException(404)
+
+
+async def _answer(utterance: str, speaker: str, deadline: float) -> dict:
+    """대화 한 턴을 SkillResponse 로. 어떤 실패도 문구로 바꾼다."""
+    try:
+        answer = await asyncio.wait_for(channel.handle(utterance, speaker), timeout=deadline)
+    except asyncio.TimeoutError:
+        LOG.w("[agent/kakao] 응답 시간 초과 — 안내로 끊음")
+        return _reply(_TIMEOUT_TEXT)
+    except Exception as ex:  # noqa: BLE001 — 메신저에서는 500 도 침묵으로 보인다
+        LOG.w(f"[agent/kakao] 처리 실패: {type(ex).__name__}")
+        return _reply(_ERROR_TEXT)
+    return _reply(answer["text"], answer.get("quick_replies"))
+
+
+async def _push(callback_url: str, utterance: str, speaker: str) -> None:
+    """답을 다 만든 뒤 콜백 주소로 보낸다.
+
+    ★ 주소는 1분 · 1회만 유효하다. 실패해도 재시도하지 않는다 — 두 번째 POST 는 어차피
+      거절되고, 사장님에게는 이미 "확인하고 있어요" 가 가 있다."""
+    payload = await _answer(utterance, speaker, CALLBACK_DEADLINE_SEC)
+    try:
+        async with httpx.AsyncClient(timeout=10.0) as client:
+            res = await client.post(callback_url, json=payload)
+        if res.status_code >= 400:
+            LOG.w(f"[agent/kakao] 콜백 전송 실패: {res.status_code}")
+    except Exception as ex:  # noqa: BLE001
+        LOG.w(f"[agent/kakao] 콜백 전송 실패: {type(ex).__name__}")
+
+
+async def _handle(body: dict, tasks: BackgroundTasks) -> dict:
+    request = body.get("userRequest") or {}
+    utterance = request.get("utterance") or ""
+    speaker = (request.get("user") or {}).get("id") or ""
+    if not speaker:
+        # 발화자를 모르면 누구의 가게인지도 모른다. 여기서 끝낸다.
+        return _reply("사용자를 확인하지 못했어요.")
+
+    # ★ 콜백이 켜져 있으면 5초 벽을 넘을 수 있다. 즉답하고 뒤에서 마저 만든다.
+    callback_url = request.get("callbackUrl")
+    # ★ "콜백을 켰는데 왜 안 되나" 를 눈으로 가릴 수 있게 남긴다. 어느 블록이 도는지도 같이 —
+    #   스킬이 폴백이 아닌 다른 블록에 붙어 있으면 콜백 설정이 그 블록에 없어 조용히 동기로 돈다.
+    LOG.i(f"[agent/kakao] 요청 — callbackUrl={'있음' if callback_url else '없음'} "
+          f"block={(request.get('block') or {}).get('name')!r}")
+    if callback_url:
+        tasks.add_task(_push, callback_url, utterance, speaker)
+        return {"version": "2.0", "useCallback": True, "data": {"text": _WAIT_TEXT}}
+
+    return await _answer(utterance, speaker, DEADLINE_SEC)
+
+
+@router.post("/webhook")
+async def webhook(
+    request: Request,
+    tasks: BackgroundTasks,
+    x_agent_secret: str | None = Header(default=None),
+):
+    """헤더로 시크릿을 받는 쪽. 스킬 설정에서 커스텀 헤더를 넣을 수 있으면 이쪽을 쓴다."""
+    body = await request.json()
+    _authorize(None, x_agent_secret, body)
+    return await _handle(body, tasks)
+
+
+@router.post("/webhook/{secret}")
+async def webhook_with_path_secret(
+    secret: str,
+    request: Request,
+    tasks: BackgroundTasks,
+    x_agent_secret: str | None = Header(default=None),
+):
+    """헤더를 못 넣는 경우의 대안.
+
+    ★ 최후 수단이다 — 경로는 액세스 로그·앞단 프록시에 남는다. 헤더를 쓸 수 있으면 위를 쓴다."""
+    body = await request.json()
+    _authorize(secret, x_agent_secret, body)
+    return await _handle(body, tasks)

solution/backend/router/v1/site/post.py

@@ -14,6 +14,7 @@
   "바로 발행" — 수정 없이 그대로 승인). PUT(수정)은 저장만 하고 자동으로 승인하지 않는다 —
   승인은 이 두 경로 중 하나를 명시적으로 눌러야 한다.
 """
+import html
 from datetime import date
 from uuid import UUID
 
@@ -30,18 +31,27 @@ from services.post_service import PostService
 router = APIRouter(prefix="/v1/site/post", tags=["Site"])
 owner_router = APIRouter(prefix="/v1/place/{place_id}/post", tags=["Site"])
 
+_REDIRECT_DELAY_SEC = 5
+
 _PAGE = """<!doctype html><html lang="ko"><head><meta charset="utf-8">
 <meta name="viewport" content="width=device-width,initial-scale=1"><meta name="robots" content="noindex">
-<title>{title}</title><style>
+{redirect}<title>{title}</title><style>
 body{{margin:0;background:#f6f5f1;color:#1b1a15;font:17px/1.7 -apple-system,'Apple SD Gothic Neo','Noto Sans KR',sans-serif}}
 .wrap{{max-width:34rem;margin:0 auto;padding:40px 20px}}
 h1{{font-size:21px;margin:0 0 6px}} p{{margin:0 0 14px}}
-.meta{{color:#6b7269;font-size:14px}}
+.meta{{color:#6b7269;font-size:14px}} a{{color:#1b1a15}}
 </style></head><body><div class="wrap">{content}</div></body></html>"""
 
 
-def _page(title: str, content: str) -> HTMLResponse:
+def _page(title: str, content: str, *, redirect_url: str | None = None) -> HTMLResponse:
-    return HTMLResponse(_PAGE.format(title=title, content=content))
+    # ★ redirect_url 은 항상 서버가 site_payload.publish_url() 로 만든 값(고정 오리진 +
+    #   slugify 통과 슬러그)이라 사용자 입력이 아니지만, HTML 속성에 그대로 꽂는 자리라
+    #   escape 를 걸어 둔다 — 이 함수가 나중에 다른 값을 받게 되더라도 안전하게.
+    redirect = (
+        f'<meta http-equiv="refresh" content="{_REDIRECT_DELAY_SEC};url={html.escape(redirect_url, quote=True)}">'
+        if redirect_url else ""
+    )
+    return HTMLResponse(_PAGE.format(title=title, content=content, redirect=redirect))
 
 
 def _expired_page() -> HTMLResponse:
@@ -56,7 +66,20 @@ async def approve_page(t: str = Query(min_length=8, max_length=200), service: Po
     result = await service.decide(t, skip=False)
     if not result["success"]:
         return _expired_page()
-    return _page("올렸습니다", f"<h1>{result['message']}</h1><p class='meta'>사이트에 반영되기까지 몇 분 걸립니다.</p>")
+    redirect_url = result.get("redirect_url")
+    # ★ 재발행은 몇 분 걸린다(BUILD 잡) — 5초 뒤에 이 글이 이미 반영돼 있다는 보장은 없다.
+    #   그래도 "어디로 가면 보이는지" 를 알려주는 게 사장님 입장에서 "눌렀는데 어디 갔지" 보다
+    #   낫다(2026-09-22, 사장님 지시). 링크 자체는 안내 문구에도 남겨 자동 이동을 못 믿어도 되게 한다.
+    extra = (
+        f"<p class='meta'>{_REDIRECT_DELAY_SEC}초 뒤 자동으로 이동합니다. "
+        f"바로 가려면 <a href='{html.escape(redirect_url, quote=True)}'>여기</a>를 눌러주세요.</p>"
+        if redirect_url else ""
+    )
+    return _page(
+        "올렸습니다",
+        f"<h1>{result['message']}</h1><p class='meta'>사이트에 반영되기까지 몇 분 걸립니다.</p>{extra}",
+        redirect_url=redirect_url,
+    )
 
 
 @owner_router.get(path="", response_model=Res_MyPosts, summary="이번 달(또는 고른 달) 생성된 글 목록")

solution/backend/scheduler/__init__.py

@@ -54,15 +54,17 @@ def start_scheduler():
         _scheduler.add_job(run_scheduled_check, "interval", minutes=10,
                            id="search-console", max_instances=1, coalesce=True)
 
-    from scheduler.jobs import sweep_alert_outbox, sweep_blog_drafts, sweep_blog_mail, sweep_queue_health
+    from scheduler.jobs import sweep_alert_outbox, sweep_blog_mail, sweep_queue_health
     _scheduler.add_job(sweep_alert_outbox, "interval", minutes=1,
                        id="alert-outbox", max_instances=1, coalesce=True)
     _scheduler.add_job(sweep_queue_health, "interval", minutes=5,
                        id="queue-health", max_instances=1, coalesce=True)
     # 미니 블로그 — 새벽에 재고를 채우고, 아침에 검수 통과분을 보낸다(docs/MINI_BLOG.md).
     # LLM 키나 메일 설정이 없으면 두 잡 모두 아무 일도 안 하고 돌아온다.
-    _scheduler.add_job(sweep_blog_drafts, "cron", hour=4, minute=10,
+    # 자동 생성은 잠시 끈다 — 사장님이 빌더에서 '생성'을 눌러야 만들어지는 흐름으로 간다(2026-09-23).
-                       id="blog-drafts", max_instances=1, coalesce=True)
+    # 되살리려면 위 import 에 sweep_blog_drafts 를 다시 넣고 아래 두 줄 주석을 푼다.
+    # _scheduler.add_job(sweep_blog_drafts, "cron", hour=4, minute=10,
+    #                    id="blog-drafts", max_instances=1, coalesce=True)
     _scheduler.add_job(sweep_blog_mail, "cron", hour=9, minute=0,
                        id="blog-mail", max_instances=1, coalesce=True)
     _scheduler.start()

solution/backend/services/agent/channel.py

@@ -0,0 +1,288 @@
+"""메신저 대화 한 턴 — 신원 · 가게 고르기 · 확인 이어받기.
+
+★★ **카카오를 모른다.** `version: "2.0"` · `simpleText` 같은 형식은 한 글자도 여기 없다.
+   그건 `router/v1/agent/kakao_bot.py` 안에서 끝난다 — 새면 다른 채널을 붙일 때 전부
+   걷어내야 하고, 알림톡 어댑터에 건 것과 같은 규칙이다.
+
+★ 빌더 화면과 무엇이 다른가 — 셋뿐이다.
+    1. 로그인 토큰이 없다        → 연결된 발화자 키로 사장님을 찾는다
+    2. place_id 가 URL 에 없다   → 대화에서 고르고 기억한다
+    3. 확인을 되돌려 줄 프론트가 없다 → 무엇을 물었는지 서버가 들고 있는다
+  나머지(도구·등급·게이트)는 `runtime.chat()` 그대로다.
+"""
+
+import re
+import uuid
+from datetime import datetime, timedelta, timezone
+
+from sqlalchemy import select
+
+from common.database.db_session_manager import DB_SESSION_MNG
+from common.database.model.models import owner_kakao_links as Link
+from common.database.model.models import users
+from common.enums import DBWRType, ErrorType, KakaoLinkStatus
+from common.models.gmodel import UserInfo
+from crud.place_crud import PlaceCRUD
+from crud.site_crud import SiteCRUD
+from crud.job_crud import JobQueue
+from common.enums import SiteStatus
+from common.models.gmodel import PageParams
+from services.site_service import SiteService
+from services import kakao_link_service as link_service
+from services.agent import runtime
+from services.agent.tools import REGISTRY
+from services.kakao_link_service import KakaoLinkError
+
+# 연결 코드 모양(kakao_link_service._CODE_ALPHABET 과 같은 글자 집합).
+CODE_PATTERN = re.compile(r"[ABCDEFGHJKMNPQRSTUVWXYZ23456789]{6}")
+
+# 확인 대기 수명. ★ 이게 없으면 한참 뒤의 "네" 한 마디에 묵은 발행이 실행된다.
+PENDING_MINUTES = 3
+
+# ★ 바로가기 라벨과 '예' 로 읽는 말이 어긋나면 **눌러도 안 먹는다** — 사장님은 버튼이
+#   고장난 줄 안다. 라벨을 상수로 두고 _YES 가 그것을 포함하게 묶는다.
+CONFIRM_LABEL = "네, 해주세요"
+PUBLISH_LABEL = "네, 발행해주세요"
+DECLINE_LABEL = "아니요"
+
+_YES = {CONFIRM_LABEL, PUBLISH_LABEL, "네", "예", "응", "그래", "네 해주세요", "해주세요", "좋아", "ㅇㅇ", "확인"}
+_NO = {DECLINE_LABEL, "아니", "아니오", "안할래", "취소", "나중에", "ㄴㄴ"}
+
+# 언제든 목록으로 돌아오는 말. ★ LLM 을 부르지 않는다 — 목록 보기에 돈을 쓸 이유가 없고,
+#   "지금 어느 가게냐" 는 대화가 막혔을 때 가장 먼저 찾는 길이라 늘 통해야 한다.
+_LIST_WORDS = {
+    "목록", "가게 목록", "사이트 목록", "내 사이트", "홈페이지 목록",
+    "가게 바꿔줘", "가게 변경", "다른 가게", "사이트 바꿔줘", "사이트 변경",
+}
+
+
+def _now():
+    return datetime.now(timezone.utc)
+
+
+def _say(text: str, quick: list[str] | None = None) -> dict:
+    """채널이 모르는 모양으로 답한다 — 문구와 바로가기 목록뿐이다."""
+    return {"text": text, "quick_replies": quick or []}
+
+
+async def _user_info(user_id) -> UserInfo | None:
+    """user_id → UserInfo. ★ 토큰을 발급하지 않는다.
+
+    프로세스 안에서 쓸 객체만 만든다 — 카톡 경로에서 JWT 가 나오면 그게 곧 권한 탈취
+    경로다(docs/AGENT.md)."""
+
+    async def run(s):
+        row = (await s.execute(select(users).where(users.user_id == user_id, users.deleted.is_(False)))).scalars().first()
+        return ErrorType.SUCCESS, row
+
+    # ★ execute_lambda 는 람다 반환값을 **그대로** 준다. CRUD 관례(ErrorType, 값)를 따라
+    #   우리 람다도 같은 모양으로 돌려준다 — 안 맞추면 여기서 TypeError 로 조용히 죽는다.
+    err, row = await DB_SESSION_MNG.execute_lambda(users.DBType(), DBWRType.DB_READ.value, run)
+    if err != ErrorType.SUCCESS or row is None:
+        return None
+    return UserInfo(user_id=str(row.user_id), id=row.id, role=row.role, token_version=row.token_version)
+
+
+async def _link_row(channel_user_key: str):
+    async def run(s):
+        row = (
+            await s.execute(
+                select(Link).where(
+                    Link.channel_user_key == channel_user_key,
+                    Link.deleted.is_(False),
+                    Link.status == KakaoLinkStatus.LINKED.value,
+                )
+            )
+        ).scalars().first()
+        return ErrorType.SUCCESS, row
+
+    _err, row = await DB_SESSION_MNG.execute_lambda(Link.DBType(), DBWRType.DB_READ.value, run)
+    return row
+
+
+async def _update_link(channel_user_key: str, **values):
+    async def run(s):
+        row = (
+            await s.execute(
+                select(Link).where(
+                    Link.channel_user_key == channel_user_key,
+                    Link.deleted.is_(False),
+                    Link.status == KakaoLinkStatus.LINKED.value,
+                )
+            )
+        ).scalars().first()
+        if row is None:
+            return None
+        for name, value in values.items():
+            setattr(row, name, value)
+        return row
+
+    await DB_SESSION_MNG.execute_lambda_write(Link.DBType(), run)
+
+
+async def _clear_pending(key):
+    await _update_link(key, pending_tool=None, pending_args=None, pending_expires_at=None)
+
+
+async def _sites(user: UserInfo) -> list:
+    """사장님의 가게 + 그 사이트 상태를 한 번에.
+
+    ★ 사업장 목록이 아니라 **사이트 목록**을 쓴다. 대화에서 사장님이 알아야 하는 것은
+      "가게가 있다" 가 아니라 "발행돼 있나 · 주소가 뭔가" 다 — `/sites` 화면이 같은 이유로
+      `list_my_sites` 를 쓴다."""
+    service = SiteService(SiteCRUD(), PlaceCRUD(), JobQueue())
+    res = await service.list_my_sites(user, PageParams(page=1, size=20))
+    return list(res.sites or [])
+
+
+def _line(row) -> str:
+    """목록 한 줄. ★ 발행 여부를 같이 말한다 — 안 그러면 사장님은 고친 것이 손님에게
+    보이는 줄 안다."""
+    if row.status == SiteStatus.PUBLISHED and row.published_at:
+        when = row.published_at.strftime("%m월 %d일")
+        return f"· {row.name} — {when} 발행"
+    return f"· {row.name} — 아직 발행 전"
+
+
+def _list_reply(rows: list, head: str) -> dict:
+    body = "\n".join(_line(r) for r in rows[:10])
+    more = f"\n(그 밖에 {len(rows) - 10}곳 더)" if len(rows) > 10 else ""
+    tail = "\n\n어느 가게 이야기일까요?" if len(rows) > 1 else ""
+    return _say(f"{head}\n{body}{more}{tail}", [r.name for r in rows[:10]] if len(rows) > 1 else [])
+
+
+async def _pick_place(user: UserInfo, row, utterance: str):
+    """어느 가게 이야기인지 정한다.
+
+    ★ 여럿인데 안 정해졌으면 **되묻는다.** 임의로 첫 가게를 고르면, 사장님은 엉뚱한 가게를
+      고쳐 놓고도 그 사실을 모른다 — 화면과 달리 대화에는 "지금 보고 있는 가게" 가 없다.
+
+    반환: (place_id, 되물을 답 or None)"""
+    rows = await _sites(user)
+    if not rows:
+        return None, _say("아직 등록된 가게가 없어요. 홈페이지를 먼저 만들어 주세요.")
+
+    # ★ 언제든 목록으로 돌아올 수 있어야 한다. 대화가 막혔을 때 처음 찾는 길이다.
+    if utterance in _LIST_WORDS:
+        await _update_link(row.channel_user_key, current_place_id=None,
+                           pending_tool=None, pending_args=None, pending_expires_at=None)
+        return None, _list_reply(rows, "관리 중인 홈페이지입니다.")
+
+    # 바로가기를 눌렀거나 가게 이름을 그대로 말한 경우 — 그 가게로 맞춘다.
+    chosen = {r.name.strip(): r for r in rows}.get(utterance.strip())
+    if chosen is not None:
+        await _update_link(row.channel_user_key, current_place_id=chosen.place_id,
+                           pending_tool=None, pending_args=None, pending_expires_at=None)
+        return None, _say(f"'{chosen.name}' 으로 맞췄습니다. 무엇을 도와드릴까요?\n"
+                          f"예) 체크인 시간 3시로 바꿔줘")
+
+    if len(rows) == 1:
+        if row.current_place_id != rows[0].place_id:
+            await _update_link(row.channel_user_key, current_place_id=rows[0].place_id)
+        return str(rows[0].place_id), None
+
+    if row.current_place_id is not None:
+        return str(row.current_place_id), None
+
+    return None, _list_reply(rows, "관리 중인 홈페이지입니다.")
+
+
+async def handle(utterance: str, channel_user_key: str) -> dict:
+    """대화 한 턴. 예외를 던지지 않는다 — 메신저에서는 500 도 침묵으로 보인다."""
+    utterance = (utterance or "").strip()
+    if not utterance:
+        return _say("무엇을 도와드릴까요?")
+
+    row = await _link_row(channel_user_key)
+
+    # ── 아직 연결되지 않은 발화자 ─────────────────────────────────────────
+    if row is None:
+        found = CODE_PATTERN.fullmatch(utterance.upper())
+        if not found:
+            return _say("먼저 홈페이지 관리자 화면의 [내 사이트]에서 카카오톡 연결 코드를 받아 보내 주세요.")
+        try:
+            user_id = await link_service.redeem(utterance, channel_user_key)
+        except KakaoLinkError:
+            # ★ 없는 코드·만료·시도 초과를 구분해 답하지 않는다(kakao_link_service 주석).
+            return _say("코드가 맞지 않거나 시간이 지났어요. 새 코드를 받아 다시 보내 주세요.")
+
+        # ★ 연결만 알리고 끝내지 않는다. 사장님은 **어느 홈페이지를 다루는 대화인지** 모른 채
+        #   말을 걸게 되고, 가게가 둘 이상이면 첫 마디부터 되묻기에 걸린다.
+        user = await _user_info(user_id)
+        rows = await _sites(user) if user else []
+        if not rows:
+            return _say("연결됐습니다. 아직 등록된 가게가 없어요 — 홈페이지를 먼저 만들어 주세요.")
+        if len(rows) == 1:
+            await _update_link(channel_user_key, current_place_id=rows[0].place_id)
+            return _say(
+                f"연결됐습니다. '{rows[0].name}' 홈페이지를 여기서 고칠 수 있어요.\n"
+                f"{_line(rows[0])}\n\n예) 체크인 시간 3시로 바꿔줘"
+            )
+        return _list_reply(rows, "연결됐습니다. 관리 중인 홈페이지입니다.")
+
+    user = await _user_info(row.user_id)
+    if user is None:
+        return _say("계정을 찾지 못했어요. 관리자 화면에서 다시 연결해 주세요.")
+
+    # ★ 이미 연결된 사람이 코드를 또 보내는 일이 실제로 있었다(2026-09-22). 그대로 두면
+    #   6자리가 그냥 발화로 모델에 넘어가 유료 호출 + 대기만 쌓인다 — 여기서 끊는다.
+    if CODE_PATTERN.fullmatch(utterance.upper()):
+        return _say("이미 연결되어 있어요. 바로 말씀하시면 됩니다.\n예) 체크인 시간 3시로 바꿔줘")
+
+    # ── 확인 이어받기 ────────────────────────────────────────────────────
+    pending = None
+    if row.pending_tool and row.pending_expires_at and row.pending_expires_at > _now():
+        pending = {"tool": row.pending_tool, "args": row.pending_args or {}}
+    elif row.pending_tool:
+        # 만료. 조용히 흘리지 않고 치운다 — 남아 있으면 다음 "네" 가 그걸 집는다.
+        await _clear_pending(channel_user_key)
+
+    if pending is not None:
+        if utterance in _YES:
+            await _clear_pending(channel_user_key)
+            result = await runtime.chat(user, str(row.current_place_id), "", confirm=pending)
+            return _say(result["reply"])
+        if utterance in _NO:
+            await _clear_pending(channel_user_key)
+            return _say("알겠습니다. 그대로 두겠습니다.")
+        # 다른 말을 했으면 그 말이 우선이다. 묵은 확인을 들고 있지 않는다.
+        await _clear_pending(channel_user_key)
+
+    # ── 가게 고르기 ──────────────────────────────────────────────────────
+    place_id, ask = await _pick_place(user, row, utterance)
+    if ask is not None:
+        return ask
+
+    # ── 도구 ─────────────────────────────────────────────────────────────
+    try:
+        result = await runtime.chat(user, place_id, utterance)
+    except runtime.AgentError as ex:
+        return _say(_ERRORS.get(str(ex), "지금은 처리할 수 없어요. 잠시 뒤 다시 말씀해 주세요."))
+
+    if result.get("needs_confirm") and result.get("tool"):
+        await _update_link(
+            channel_user_key,
+            pending_tool=result["tool"],
+            pending_args=result.get("args") or {},
+            pending_expires_at=_now() + timedelta(minutes=PENDING_MINUTES),
+        )
+        return _say(result["reply"], [CONFIRM_LABEL, DECLINE_LABEL])
+
+    # 값을 고쳤으면 재발행을 바로 누를 수 있게 바로가기를 붙인다 — 도구가 이미 그렇게 묻는다.
+    quick = [PUBLISH_LABEL, DECLINE_LABEL] if result.get("done") and result.get("tool") != REGISTRY["publish"].name else []
+    if quick:
+        await _update_link(
+            channel_user_key,
+            pending_tool="publish",
+            pending_args={},
+            pending_expires_at=_now() + timedelta(minutes=PENDING_MINUTES),
+        )
+    return _say(result["reply"], quick)
+
+
+_ERRORS = {
+    "PLACE_NOT_FOUND": "그 가게를 찾지 못했어요.",
+    "AGENT_NOT_CONFIGURED": "지금은 대화 기능이 꺼져 있어요.",
+    "AGENT_MESSAGE_TOO_LONG": "말씀이 조금 길어요. 짧게 나눠서 말씀해 주세요.",
+    "AGENT_CALL_FAILED": "지금은 처리할 수 없어요. 잠시 뒤 다시 말씀해 주세요.",
+}

solution/backend/services/agent/runtime.py

@@ -45,11 +45,10 @@ class AgentError(RuntimeError):
 def is_configured() -> bool:
     """대화창을 열 수 있나 — 스위치와 LLM 키를 **둘 다** 본다.
 
-    ★ `AGENT_CHAT_ENABLED` 가 기본 꺼짐이다(config/agent_config). 카카오톡 채널이 준비되기
+    ★ 스위치(`AGENT_CHAT_ENABLED`)와 키를 **둘 다** 보는 이유: 키만 보면 "잠시 닫아 두기" 를
-      전에는 대화창을 띄우지 않는다 — 코드는 다 있지만 사장님 입장에서는 어디에도 닿지 않는
+      키를 지워서 해야 하는데 그러면 소개문·사진분류까지 같이 꺼진다. 스위치만 보면
-      입구이고, 열려 있으면 "되는 기능" 으로 오해한다.
+      키 없는 환경에서 **눌러도 안 되는 입구**가 생긴다.
-      이건 Threads 카드처럼 '자리는 두고 버튼만 죽이는' 경우와 다르다. 저쪽은 사장님이
+      실제로 2026-09-21 에 카카오 채널 보류로 한 번 닫았고, 채널 인증이 끝나 다시 열었다."""
-      **곧 쓸 수 있는** 기능이라 존재를 알려야 했고, 이쪽은 아직 제품이 아니다."""
     return config.chat_enabled() and provider.active().is_configured()
 
 
@@ -72,14 +71,18 @@ async def _context_facts(user: UserInfo, place_id: str, place) -> list[dict]:
     res = await FactService(FactCRUD(), PlaceCRUD()).list_facts(user, place_id, publishable_only=True)
     schema = get_schema(PlaceCategory(place.category))
     out = []
-    for f in (res.facts or [])[:60]:
+    for f in (res.facts or []):
         spec = schema.get(f.key)
         if spec and spec.scope == "place" and (f.value or "").strip():
-            out.append({"key": f.key, "label": spec.label, "value": f.value})
+            # ★ label 은 싣지 않는다 — 아래 '항목 목록' 에 이미 key↔label 이 있다.
-    return out
+            #   같은 표를 두 번 보내면 프롬프트만 커지고 모델이 얻는 것은 없다.
+            out.append({f.key: f.value})
+    # ★ 상한을 둔다. 실측(2026-09-22): 필드 43 + fact 수십 개가 실린 프롬프트가 5초 벽을
+    #   넘겼다. 무한정 싣지 않는다 — 대화 한 턴에 필요한 맥락은 그렇게 많지 않다.
+    return out[:30]
 
 
-async def _choose(place, fields, facts, site_line, message) -> dict:
+async def _choose(place, fields, facts, message) -> dict:
     """LLM 한 번. 고른 도구 이름과 인자만 받는다."""
     active = provider.active()
     async with httpx.AsyncClient(timeout=REQUEST_TIMEOUT) as client:
@@ -91,7 +94,6 @@ async def _choose(place, fields, facts, site_line, message) -> dict:
                 tools=registry.describe(),
                 fields=fields,
                 facts=facts,
-                site={"요약": site_line},
                 message=message,
             ),
             response_schema=prompt.RESPONSE_SCHEMA,
@@ -127,10 +129,11 @@ async def chat(user: UserInfo, place_id: str, message: str, confirm: dict | None
 
     fields = registry.fields_of(place)
     facts = await _context_facts(user, place_id, place)
-    site_line = await registry.REGISTRY["get_site_status"].run(ctx, {})
+    # ★ 사이트 상태는 프롬프트에 싣지 않는다. 그 한 줄 때문에 매 턴 사이트 조회 + 슬러그 계산이
+    #   돌았고, 정작 모델이 필요할 때는 `get_site_status` 도구를 부르면 된다.
 
     try:
-        choice = await _choose(place, fields, facts, site_line, message)
+        choice = await _choose(place, fields, facts, message)
     except LlmError as ex:
         LOG.w(f"[agent] 도구 선택 실패: {type(ex).__name__}")
         raise AgentError("AGENT_CALL_FAILED") from ex

solution/backend/services/blog_jobs.py

@@ -3,7 +3,7 @@
 ★ 잡은 '대상을 고르는 것'까지만 하고 실제 일은 서비스가 한다(scheduler/jobs.py 규약).
 ★ 한 번에 BATCH_SIZE 건씩 만든다. 한 달치를 한 호출로 뽑으면 앞 회차 주제를 프롬프트에
   못 넣어 중복이 막히지 않는다.
-★ 팀 사전검수 없음 — 금칙 필터(blog_service.filter_drafts)를 통과하면 바로 REVIEWED 로
+★ 팀 사전검수 없음 — 금칙 필터(blog_service.is_publishable_body)를 통과하면 바로 REVIEWED 로
   쌓이고, send_reviewed() 가 업장당 하루 한 통씩 그대로 사장님에게 보낸다.
 ★ 글마다 scheduled_date(KST) 를 하나씩 배정한다 — "언제 만들어졌나"만 있고 "언제 낼
   것인가"가 없으면 달력 화면이 근거 없는 날짜를 지어내야 한다(2026-09-17).
@@ -73,11 +73,14 @@ async def _pending_count(place_id) -> int:
     return len(result.all()) if result is not None else 0
 
 
-async def _generate_for_place(place) -> int:
+async def _compose_for_dates(place, dates: list[date]) -> list[dict]:
-    """업장 하나. 재고가 이미 REFILL_BELOW 이상이면 아무것도 안 만든다(만든 수 0)."""
+    """날짜마다 그 날짜에 맞는 소재(blog_service.materials(snapshot, d))로 한 편씩 만든다 —
-    if await _pending_count(place.place_id) >= REFILL_BELOW:
+    세 생성 경로(자동·구간·개별)가 같이 쓴다. 저장은 부르는 쪽이 한다.
-        return 0
 
+    ★ 날짜를 먼저 정하고 소재를 고른다(2026-09-23). 예전에는 소재 목록을 순서대로 뽑아 날짜에
+      차례로 붙여서, 글 내용이 배정된 날짜와 무관했다.
+    ★ 그 날짜에 맞는 소재가 없으면 그 날짜만 비워 두고 다음 날짜로 간다 — 뒤 날짜엔 축제가 걸릴 수 있다.
+    ★ LLM 이 없거나 실패하면(None) 그 자리에서 멈춘다 — 날짜마다 소재를 전부 돌며 헛호출하지 않는다."""
     used = await DB_SESSION_MNG.execute_lambda(
         place_posts.DBType(), DBWRType.DB_READ.value,
         lambda s, pid=place.place_id: _crud.used_topic_keys(s, pid),
@@ -87,39 +90,47 @@ async def _generate_for_place(place) -> int:
     region = site_payload.region_label(place.road_address, place.address)
 
     rows = []
-    for kind, key, material in blog_service.materials(snapshot):
+    for target in dates:
-        if len(rows) >= BATCH_SIZE:
+        for kind, key, material in blog_service.materials(snapshot, target):
+            if key in used_set:
+                continue
+            generated = await blog_service.generate_one(
+                place_name=place.name, region=region, topic_kind=kind, material=material,
+                used_topics=sorted(used_set), place_category=place.category, post_date=target,
+            )
+            if not generated:
+                return rows
+            body, model = generated
+            ok, reason = blog_service.is_publishable_body(body)
+            used_set.add(key)  # 버린 주제도 이번 회차에서 다시 고르지 않는다
+            if not ok:
+                LOG.i(f"[blog] place={place.place_id} {target} 버림 — {reason}")
+                continue
+            rows.append({
+                "place_id": place.place_id, "body": body, "topic_kind": kind, "topic_key": key,
+                "scheduled_date": target, "generation_meta": {"model": model},
+                "status": PostStatus.REVIEWED.value,  # 금칙 필터를 이미 통과했다 — 팀 사전검수 없음
+            })
             break
-        if key in used_set:
+    return rows
-            continue
-        generated = await blog_service.generate_one(
-            place_name=place.name, region=region, topic_kind=kind,
-            material=material, used_topics=sorted(used_set), place_category=place.category,
-        )
-        if not generated:
-            continue
-        body, model = generated
-        rows.append({
-            "place_id": place.place_id, "body": body, "topic_kind": kind, "topic_key": key,
-            "generation_meta": {"model": model},
-        })
-        used_set.add(key)
 
-    kept, dropped = blog_service.filter_drafts(rows)
+
-    if kept:
+async def _generate_for_place(place) -> int:
-        latest = await DB_SESSION_MNG.execute_lambda(
+    """업장 하나. 재고가 이미 REFILL_BELOW 이상이면 아무것도 안 만든다(만든 수 0)."""
-            place_posts.DBType(), DBWRType.DB_READ.value,
+    if await _pending_count(place.place_id) >= REFILL_BELOW:
-            lambda s, pid=place.place_id: _crud.max_scheduled_date(s, pid),
+        return 0
-        )
+
-        next_date = max(latest + timedelta(days=1), _today_kst()) if latest else _today_kst()
+    latest = await DB_SESSION_MNG.execute_lambda(
-        for offset, row in enumerate(kept):
+        place_posts.DBType(), DBWRType.DB_READ.value,
-            row["scheduled_date"] = next_date + timedelta(days=offset)
+        lambda s, pid=place.place_id: _crud.max_scheduled_date(s, pid),
+    )
+    next_date = max(latest + timedelta(days=1), _today_kst()) if latest else _today_kst()
+    rows = await _compose_for_dates(place, [next_date + timedelta(days=i) for i in range(BATCH_SIZE)])
+    if rows:
         await DB_SESSION_MNG.execute_lambda_run(
-            [place_posts.DBType()], [lambda s, r=kept: _crud.add_many(s, r)],
+            [place_posts.DBType()], [lambda s, r=rows: _crud.add_many(s, r)],
         )
-    if dropped:
+    return len(rows)
-        LOG.i(f"[blog] place={place.place_id} 버린 {len(dropped)}건 — {dropped[0][1]}")
-    return len(kept)
 
 
 async def generate_drafts() -> int:
@@ -135,8 +146,8 @@ async def generate_range(place_id: str, start_date: date, end_date: date) -> dic
     (2026-09-17, 사장님 지시: "지금 생성하기에서 시작이랑 끝 날짜를 정해야하지 않을까").
     재고 상한(REFILL_BELOW)을 안 본다 — 개별 생성과 같은 이유로, 직접 고른 구간에
     상한 로직이 끼어들 자리가 아니다. 이미 글이 있는 날짜는 LLM 을 부르지 않고 건너뛴다 —
-    매번 새로 만들고 유니크 충돌로 버리면 호출만 낭비된다. 소재가 떨어지면(구간 안에서
+    매번 새로 만들고 유니크 충돌로 버리면 호출만 낭비된다. 그 날짜에 맞는 소재가 없으면
-    더 만들 topic_key 가 없으면) 그 자리에서 멈춘다 — 못 채운 나머지는 그대로 빈 날짜로 남는다."""
+    그 날짜는 빈 날짜로 남는다(_compose_for_dates)."""
     place = None
     for p, _user in await _published_places():
         if str(p.place_id) == str(place_id):
@@ -157,38 +168,7 @@ async def generate_range(place_id: str, start_date: date, end_date: date) -> dic
     if not empty_dates:
         return {"requested": requested, "created": 0}
 
-    used = await DB_SESSION_MNG.execute_lambda(
+    rows = await _compose_for_dates(place, empty_dates)
-        place_posts.DBType(), DBWRType.DB_READ.value,
-        lambda s, pid=place.place_id: _crud.used_topic_keys(s, pid),
-    )
-    used_set = set(used or [])
-    snapshot = await build_snapshot(place)
-    region = site_payload.region_label(place.road_address, place.address)
-
-    rows = []
-    for kind, key, material in blog_service.materials(snapshot):
-        if len(rows) >= len(empty_dates):
-            break
-        if key in used_set:
-            continue
-        generated = await blog_service.generate_one(
-            place_name=place.name, region=region, topic_kind=kind,
-            material=material, used_topics=sorted(used_set), place_category=place.category,
-        )
-        if not generated:
-            continue
-        body, model = generated
-        ok, _reason = blog_service.is_publishable_body(body)
-        used_set.add(key)
-        if not ok:
-            continue
-        rows.append({
-            "place_id": place.place_id, "body": body, "topic_kind": kind, "topic_key": key,
-            "generation_meta": {"model": model}, "status": PostStatus.REVIEWED.value,
-        })
-
-    for row, d in zip(rows, empty_dates):
-        row["scheduled_date"] = d
     if rows:
         await DB_SESSION_MNG.execute_lambda_run(
             [place_posts.DBType()], [lambda s, r=rows: _crud.add_many(s, r)],
@@ -208,39 +188,12 @@ async def generate_one_for_date(place_id: str, target_date: date) -> dict | None
     if place is None:
         return None
 
-    used = await DB_SESSION_MNG.execute_lambda(
+    rows = await _compose_for_dates(place, [target_date])
-        place_posts.DBType(), DBWRType.DB_READ.value,
+    if not rows:
-        lambda s, pid=place.place_id: _crud.used_topic_keys(s, pid),
+        return None
-    )
+    return await DB_SESSION_MNG.execute_lambda_write(
-    used_set = set(used or [])
+        place_posts.DBType(), lambda s, r=rows[0]: _crud.add_one(s, r),
-    snapshot = await build_snapshot(place)
+    )  # None 이면 그 날짜(또는 주제)가 이미 차 있었다 — 다시 시도하지 않는다
-    region = site_payload.region_label(place.road_address, place.address)
-
-    for kind, key, material in blog_service.materials(snapshot):
-        if key in used_set:
-            continue
-        generated = await blog_service.generate_one(
-            place_name=place.name, region=region, topic_kind=kind,
-            material=material, used_topics=sorted(used_set), place_category=place.category,
-        )
-        if not generated:
-            continue
-        body, model = generated
-        ok, _reason = blog_service.is_publishable_body(body)
-        if not ok:
-            used_set.add(key)  # 이 주제는 이번 시도에서 다시 고르지 않는다
-            continue
-
-        row = {
-            "place_id": place.place_id, "body": body, "topic_kind": kind, "topic_key": key,
-            "scheduled_date": target_date, "generation_meta": {"model": model},
-            "status": PostStatus.REVIEWED.value,  # 금칙 필터를 이미 통과했다 — 벌크 경로(filter_drafts)와 동일
-        }
-        inserted = await DB_SESSION_MNG.execute_lambda_write(
-            place_posts.DBType(), lambda s, r=row: _crud.add_one(s, r),
-        )
-        return inserted  # None 이면 그 날짜(또는 주제)가 이미 차 있었다 — 다시 시도하지 않는다
-    return None
 
 
 def _mail_body(*, place_name: str, post, user, origin: str, approve_token: str) -> str:

solution/backend/services/blog_service.py

@@ -8,9 +8,9 @@
 import hashlib
 import re
 import secrets
-from datetime import datetime, timedelta, timezone
+from datetime import date, datetime, timedelta, timezone
 
-from common.enums import PlaceCategory, PostStatus, PostTopicKind
+from common.enums import LocalContentType, PlaceCategory, PostStatus, PostTopicKind
 from common.logger import LOG
 
 # 본문 길이 — 회의 확정값(140~150자)에 여유를 둔다. 벗어나면 버린다.
@@ -66,12 +66,13 @@ def issue_token() -> tuple[str, str, object]:
 
 # 숙소(LODGING) 기본 갈래 규칙 — 업종별 규칙이 없을 때의 폴백이기도 하다.
 TOPIC_RULES: dict[int, str] = {
+    # ★ 게시일의 실제 날씨는 모른다(글은 며칠·몇 주 앞서 만든다) — "오늘은 비가 옵니다"라고 단정하게 두지 않는다.
     PostTopicKind.WEATHER.value:
-        "오늘의 날씨와 그 날씨에 이 숙소에서 하기 좋은 일을 한 장면으로 적는다.",
+        "소재로 주어진 날씨인 날, 이 숙소에서 하기 좋은 일을 한 장면으로 적는다. 게시일의 날씨를 단정하지 않는다.",
     PostTopicKind.FESTIVAL.value:
-        "주어진 축제 하나를 언급하고, 숙소에서 그곳까지 어떻게 가는지를 걸음 단위로 적는다.",
+        "주어진 축제 하나를 게시일 기준으로(곧 열리는지, 열리는 중인지) 언급하고, 숙소에서 그곳까지 어떻게 가는지를 걸음 단위로 적는다.",
     PostTopicKind.SEASON.value:
-        "지금 절기에 이 지역과 숙소가 어떻게 달라지는지를 적는다.",
+        "게시일 무렵 절기에 이 지역과 숙소가 어떻게 달라지는지를 적는다.",
     PostTopicKind.NEARBY.value:
         "주어진 주변 장소 하나를 손님 시선에서 적는다. 영업시간과 가격은 쓰지 않는다.",
     PostTopicKind.GUIDE.value:
@@ -102,17 +103,36 @@ _RULES = (
     "- '최고' '유일' 같은 최상급을 쓰지 않는다.\n"
     "- 손님에게 말하듯 존댓말로 적는다.\n"
     "- 아래 '이미 쓴 주제'와 겹치는 소재를 고르지 않는다.\n"
+    "- 게시일과 맞지 않는 계절·날씨·행사 이야기를 쓰지 않는다.\n"
 )
 
+_WEEKDAYS = "월화수목금토일"
+
+
+def season_term(on: date) -> str:
+    """게시일 → 절기 이름(materials 의 계절 소재와 같은 말). 달로만 가른다."""
+    return {
+        3: "봄", 4: "봄", 5: "봄",
+        6: "초여름", 7: "한여름", 8: "한여름",
+        9: "초가을", 10: "늦가을", 11: "늦가을",
+        12: "초겨울", 1: "한겨울", 2: "한겨울",
+    }[on.month]
+
+
+def _date_line(on: date) -> str:
+    return f"게시일: {on.year}년 {on.month}월 {on.day}일({_WEEKDAYS[on.weekday()]}) · {season_term(on)}\n"
+
 
 def build_prompt(*, place_name: str, region: str, topic_kind: int, material: str, used_topics: list[str],
-                  place_category: int = PlaceCategory.LODGING.value) -> str:
+                  place_category: int = PlaceCategory.LODGING.value, post_date: date | None = None) -> str:
-    """갈래 하나에 대한 프롬프트 한 벌. 프롬프트를 두 곳에 적지 않으려고 여기서만 만든다."""
+    """갈래 하나에 대한 프롬프트 한 벌. 프롬프트를 두 곳에 적지 않으려고 여기서만 만든다.
+    post_date 가 있으면 게시일을 알려 준다 — 글이 그 날짜의 계절·행사와 맞게 쓰이도록(2026-09-23)."""
     used = ", ".join(used_topics[:40]) or "없음"
     noun = _business_noun(place_category)
     rules = _topic_rules(place_category)
     return (
         f"{region}에 있는 {noun} '{place_name}'의 짧은 홍보 글을 쓴다.\n"
+        f"{_date_line(post_date) if post_date else ''}"
         f"갈래: {rules.get(topic_kind, '')}\n"
         f"소재: {material}\n"
         f"이미 쓴 주제: {used}\n\n"
@@ -146,7 +166,7 @@ def filter_drafts(rows: list[dict]) -> tuple[list[dict], list[tuple[str, str]]]:
 
 async def generate_one(*, place_name: str, region: str, topic_kind: int, material: str,
                        used_topics: list[str], place_category: int = PlaceCategory.LODGING.value,
-                       client=None) -> tuple[str, str] | None:
+                       post_date: date | None = None, client=None) -> tuple[str, str] | None:
     """(문구, 모델명) 한 쌍. LLM 이 없거나 실패하면 None — 생성 실패가 잡을 죽이지 않는다.
     모델명은 생성 이력 화면이 "어느 모델썼는지" 보여주는 데 쓴다(2026-09-17, 사장님 지시).
 
@@ -166,7 +186,8 @@ async def generate_one(*, place_name: str, region: str, topic_kind: int, materia
         return None
 
     prompt = build_prompt(place_name=place_name, region=region, topic_kind=topic_kind,
-                          material=material, used_topics=used_topics, place_category=place_category)
+                          material=material, used_topics=used_topics, place_category=place_category,
+                          post_date=post_date)
     owns = client is None
     if owns:
         import httpx
@@ -183,27 +204,68 @@ async def generate_one(*, place_name: str, region: str, topic_kind: int, materia
             await client.aclose()
 
 
-def materials(snapshot: dict) -> list[tuple[int, str, str]]:
+# 축제 글을 시작일 며칠 전부터 낼 수 있나. 끝난 축제는 내지 않는다.
-    """(갈래, topic_key, 소재). 소재가 없는 갈래는 아예 만들지 않는다 — 지어내지 않는다."""
+FESTIVAL_LEAD_DAYS = 14
-    local = snapshot.get("local") or {}
+
+# 그 달에 말이 되는 날씨만 소재로 쓴다 — 여름에 "눈인 날" 글이 나가지 않게.
+_SKIES = ("맑음", "흐림", "비", "안개")
+_SKIES_BY_MONTH = {12: ("눈",), 1: ("눈",), 2: ("눈",), 6: ("소나기",), 7: ("소나기",), 8: ("소나기",)}
+
+
+def _ymd(value) -> date | None:
+    digits = "".join(ch for ch in str(value or "") if ch.isdigit())
+    if len(digits) != 8:
+        return None
+    try:
+        return date(int(digits[:4]), int(digits[4:6]), int(digits[6:]))
+    except ValueError:
+        return None
+
+
+def materials(snapshot: dict, on: date) -> list[tuple[int, str, str]]:
+    """게시일 on 에 맞는 (갈래, topic_key, 소재). 앞에 있을수록 먼저 고른다 — 축제 → 계절 → 주변 → 날씨.
+    소재가 없는 갈래는 아예 만들지 않는다 — 지어내지 않는다.
+
+    ★ 날짜에 맞춘다(2026-09-23). 예전에는 날짜와 무관한 한 줄 목록이라, 9월 날짜에 '한겨울' 글이나
+      이미 끝난 축제 글이 붙을 수 있었다.
+      - 축제: 시작 FESTIVAL_LEAD_DAYS 일 전 ~ 끝나는 날 사이에만. 기간을 모르는 축제는 쓰지 않는다.
+      - 계절: 게시일의 절기 하나. 키에 연도를 넣어 해마다 한 번씩 다시 쓸 수 있다.
+      - 날씨: 그 달에 있을 법한 것만. 키에 연·월을 넣어 달마다 다시 쓸 수 있다.
+      - 주변 장소: 날짜와 무관해 늘 후보다.
+    ★ 스냅샷의 지역 정보는 원문 행 목록(snapshot["local"]["contents"])이다. 예전 코드는
+      site_payload 모양(local.festivals·attractions)을 읽어 축제·주변 소재가 늘 비어 있었다."""
+    contents = (snapshot.get("local") or {}).get("contents") or []
+    by_type: dict[int, list[dict]] = {}
+    for row in contents:
+        if isinstance(row, dict):
+            by_type.setdefault(row.get("content_type"), []).append(row)
     out: list[tuple[int, str, str]] = []
 
-    for festival in (local.get("festivals") or [])[:12]:
+    for row in by_type.get(LocalContentType.FESTIVAL.value, []):
-        name = (festival.get("name") or "").strip()
+        body = row.get("body") or {}
+        name = str(body.get("name") or row.get("title") or "").strip()
+        start = _ymd(body.get("eventstartdate"))
+        end = _ymd(body.get("eventenddate")) or start
+        if not name or start is None or not (start - timedelta(days=FESTIVAL_LEAD_DAYS) <= on <= end):
+            continue
+        period = f"{start:%Y.%m.%d}" + (f" ~ {end:%Y.%m.%d}" if end != start else "")
+        detail = f"{body.get('location') or ''} {str(body.get('overview') or '')[:300]}".strip()
+        out.append((PostTopicKind.FESTIVAL.value, f"festival:{start.year}:{name}"[:120],
+                    f"{name} (기간 {period}) — {detail}".strip(" —")))
+
+    term = season_term(on)
+    out.append((PostTopicKind.SEASON.value, f"season:{on.year}:{term}", term))
+
+    spots = (by_type.get(LocalContentType.ATTRACTION.value, [])
+             + by_type.get(LocalContentType.RESTAURANT.value, []))
+    for row in spots[:20]:
+        body = row.get("body") or {}
+        name = str(body.get("name") or row.get("title") or "").strip()
         if name:
-            out.append((PostTopicKind.FESTIVAL.value, f"festival:{name}",
+            detail = body.get("description") or body.get("overview") or body.get("location") or ""
-                        f"{name} — {festival.get('location') or ''} {festival.get('description') or ''}".strip()))
+            out.append((PostTopicKind.NEARBY.value, f"nearby:{name}"[:120], f"{name} — {detail}".strip(" —")))
 
-    for spot in ((local.get("attractions") or []) + (local.get("restaurants") or []))[:20]:
+    for sky in _SKIES + _SKIES_BY_MONTH.get(on.month, ()):
-        name = (spot.get("name") or "").strip()
+        out.append((PostTopicKind.WEATHER.value, f"weather:{on:%Y-%m}:{sky}", f"{sky}인 날"))
-        if name:
-            out.append((PostTopicKind.NEARBY.value, f"nearby:{name}",
-                        f"{name} — {spot.get('description') or spot.get('address') or ''}".strip()))
-
-    for sky in ("맑음", "흐림", "비", "눈", "안개", "소나기"):
-        out.append((PostTopicKind.WEATHER.value, f"weather:{sky}", f"{sky}인 날"))
-
-    for term in ("봄", "초여름", "한여름", "초가을", "늦가을", "초겨울", "한겨울"):
-        out.append((PostTopicKind.SEASON.value, f"season:{term}", term))
 
     return out

solution/backend/services/post_service.py

@@ -15,15 +15,15 @@ from crud.job_crud import JobQueue
 from crud.place_crud import PlaceCRUD
 from crud.post_crud import PostCRUD
 from common.database.db_session_manager import DB_SESSION_MNG
-from common.database.model.models import jobs as jobs_table, place_posts, places
+from common.database.model.models import jobs as jobs_table, place_posts, places, sites
-from common.enums import DBWRType, ErrorType, JobStatus, JobType, PostStatus
+from common.enums import DBWRType, ErrorType, JobStatus, JobType, PostStatus, SiteStatus
 from common.logger import LOG
 from common.models.gmodel import Res_WebPacketProtocol, UserInfo
 from common.utils.gtime import GTime
 from router.v1.site.protocol import (
     GenerationBatch, PostData, Res_GenerateNow, Res_GenerateOne, Res_GenerationHistory, Res_MyPosts,
 )
-from services import blog_jobs, blog_service, social_service
+from services import blog_jobs, blog_service, site_payload, social_service
 from services.job_service import enqueue_job
 
 EXPIRED = "처리할 수 없는 링크입니다."
@@ -76,7 +76,36 @@ class PostService:
             return {"success": True, "message": SKIPPED}
 
         await self._approve_and_publish(post_id, place_id)
-        return {"success": True, "message": APPROVED}
+        return {
+            "success": True,
+            "message": APPROVED,
+            "redirect_url": await self._blog_url(place_id),
+        }
+
+    async def _blog_url(self, place_id) -> str | None:
+        """이 업장의 발행된 사이트에서 미니 블로그가 보이는 자리. 승인 확인 화면이 몇 초
+        뒤 여기로 자동 연결한다(2026-09-22, 사장님 지시) — 사장님이 승인만 하고 실제로
+        어디에 올라갔는지 못 찾는 걸 줄인다. 사이트가 없거나 아직 미발행이면 None —
+        호출부가 자동 연결 없이 확인 문구만 보여준다."""
+        def query(session):
+            return session.execute(
+                select(places, sites)
+                .join(sites, sites.place_id == places.place_id)
+                .where(
+                    places.place_id == place_id,
+                    places.deleted == False,  # noqa: E712
+                    sites.deleted == False,  # noqa: E712
+                    sites.status == SiteStatus.PUBLISHED.value,
+                )
+                .limit(1)
+            )
+
+        result = await DB_SESSION_MNG.execute_lambda(places.DBType(), DBWRType.DB_READ.value, query)
+        row = result.first() if result is not None else None
+        if not row:
+            return None
+        place, site = row
+        return f"{site_payload.publish_url(place, site)}#blog"
 
     async def _load_place(self, user_info: UserInfo, place_id: str):
         err_type, place = await DB_SESSION_MNG.execute_lambda(

solution/backend/services/prompts/agent.py

@@ -36,7 +36,7 @@ RESPONSE_SCHEMA = {
 }
 
 
-def build_prompt(*, place_name: str, tools: list[dict], fields: list[dict], facts: list[dict], site: dict, message: str) -> str:
+def build_prompt(*, place_name: str, tools: list[dict], fields: list[dict], facts: list[dict], message: str) -> str:
     """사장님 발화 → 도구 하나.
 
     ★ 모호하면 실행하지 말고 되물으라고 명시한다. 티오더가 "유사한 메뉴가 2개 이상이면
@@ -55,14 +55,11 @@ def build_prompt(*, place_name: str, tools: list[dict], fields: list[dict], fact
 쓸 수 있는 도구:
 {json.dumps(tools, ensure_ascii=False, indent=1)}
 
-가게 정보에 쓸 수 있는 항목(set_fact 의 key 는 반드시 이 중 하나다):
+가게 정보에 쓸 수 있는 항목 — `key: 이름` (set_fact 의 key 는 반드시 이 중 하나다):
-{json.dumps(fields, ensure_ascii=False)}
+{chr(10).join(f"{f['key']}: {f['label']}" for f in fields)}
 
 지금 저장된 값:
 {json.dumps(facts, ensure_ascii=False)}
 
-사이트 상태:
-{json.dumps(site, ensure_ascii=False)}
-
 사장님 요청:
 {message}'''

solution/backend/tests/test_agent_runtime.py

@@ -227,20 +227,22 @@ def test_읽기_도구는_확인을_요구하지_않는다():
 
 # ── 보류 스위치 ─────────────────────────────────────────────────────────
 
-def test_스위치가_꺼져_있으면_키가_있어도_안_열린다(monkeypatch):
+def test_스위치와_키를_둘_다_본다(monkeypatch):
-    """★ 기본이 꺼짐이다. 카카오톡 채널이 준비되기 전에는 대화창을 띄우지 않는다 —
+    """★ 키만 보면 '잠시 닫아 두기' 를 키를 지워서 해야 하고, 그러면 소개문·사진분류까지
-    코드는 다 있지만 사장님에게는 어디에도 닿지 않는 입구다."""
+    같이 꺼진다. 스위치만 보면 키 없는 환경에서 **눌러도 안 되는 입구**가 생긴다."""
     monkeypatch.setattr(runtime.provider, "active", lambda: SimpleNamespace(is_configured=lambda: True))
     monkeypatch.delenv("AGENT_CHAT_ENABLED", raising=False)
-    assert runtime.is_configured() is False
+    assert runtime.is_configured() is True          # 기본은 켜짐(2026-09-22)
-    monkeypatch.setenv("AGENT_CHAT_ENABLED", "1")
-    assert runtime.is_configured() is True
     monkeypatch.setenv("AGENT_CHAT_ENABLED", "0")
-    assert runtime.is_configured() is False
+    assert runtime.is_configured() is False         # 스위치로 닫을 수 있다
+
+    monkeypatch.setenv("AGENT_CHAT_ENABLED", "1")
+    monkeypatch.setattr(runtime.provider, "active", lambda: SimpleNamespace(is_configured=lambda: False))
+    assert runtime.is_configured() is False         # 켜 뒀어도 키가 없으면 안 열린다
 
 
-async def test_꺼진_동안_대화_요청은_거절된다(client, auth_headers, monkeypatch, db_engine):
+async def test_닫아_두면_대화_요청은_거절된다(client, auth_headers, monkeypatch, db_engine):
-    monkeypatch.delenv("AGENT_CHAT_ENABLED", raising=False)
+    monkeypatch.setenv("AGENT_CHAT_ENABLED", "0")
     h, pid = await seed(client, auth_headers)
     res = await client.post(f"/v1/agent/chat/{pid}", headers=h, json={"message": "안녕"})
     assert res.status_code == 409

solution/backend/tests/test_blog_owner.py

@@ -140,7 +140,7 @@ async def test_deleting_a_post_frees_its_date_for_regeneration(client, db_engine
     바로 다시 생성할 수 있어야 한다."""
     from services import blog_service
 
-    async def fake_generate_one(*, place_name, region, topic_kind, material, used_topics, place_category):
+    async def fake_generate_one(*, place_name, region, topic_kind, material, used_topics, place_category, post_date=None):
         return ("새로 만든 문구입니다. " + BODY, "gemini-test-model")
 
     monkeypatch.setattr(blog_service, "generate_one", fake_generate_one)
@@ -201,7 +201,7 @@ async def test_generate_now_creates_posts_for_published_site(client, db_engine,
     사장님 지시: "지금 생성하기에서 시작이랑 끝 날짜를 정해야하지 않을까") — 발행된 사이트일 때만."""
     from services import blog_service
 
-    async def fake_generate_one(*, place_name, region, topic_kind, material, used_topics, place_category):
+    async def fake_generate_one(*, place_name, region, topic_kind, material, used_topics, place_category, post_date=None):
         return ("테스트로 만든 문구입니다. " + BODY, "gemini-test-model")
 
     monkeypatch.setattr(blog_service, "generate_one", fake_generate_one)
@@ -241,12 +241,52 @@ async def test_generate_now_creates_posts_for_published_site(client, db_engine,
     assert all(start <= d <= end for d in dates)
 
 
+async def test_generate_now_writes_each_post_for_its_own_date(client, db_engine, auth_headers, monkeypatch):
+    """글마다 배정된 날짜를 게시일로 받아 쓰고, 그 날짜의 절기 소재가 붙는다(2026-09-23,
+    사장님 지시: "날짜에 맞는 글이 생성 되도록")."""
+    from services import blog_service
+
+    calls = []
+
+    async def fake_generate_one(*, place_name, region, topic_kind, material, used_topics, place_category, post_date=None):
+        calls.append((post_date, topic_kind, material))
+        return ("테스트로 만든 문구입니다. " + BODY, "gemini-test-model")
+
+    monkeypatch.setattr(blog_service, "generate_one", fake_generate_one)
+
+    h = await auth_headers("bloggen18")
+    place_id = await _place(client, h, name="날짜맞춤펜션")
+    async with db_engine.begin() as conn:
+        await conn.execute(
+            text("INSERT INTO sites (site_id, place_id, domain, status) VALUES (:sid, :pid, :dom, :st)"),
+            {"sid": uuid.uuid4(), "pid": place_id, "dom": f"blog-test-{uuid.uuid4().hex[:8]}", "st": SiteStatus.PUBLISHED.value},
+        )
+    winter = date(date.today().year + 1, 1, 10)
+
+    res = await client.post(
+        f"/v1/place/{place_id}/post/generate", headers=h,
+        params={"start": winter.isoformat(), "end": winter.isoformat()},
+    )
+
+    assert res.json()["created"] == 1
+    assert calls[0][0] == winter
+    # 지역 소재가 없는 업장이라 첫 후보는 그 날짜의 절기다.
+    assert calls[0][1] == PostTopicKind.SEASON.value
+    assert calls[0][2] == "한겨울"
+    async with db_engine.begin() as conn:
+        row = (await conn.execute(
+            text("SELECT scheduled_date, topic_key FROM place_posts WHERE place_id = :pid"), {"pid": place_id},
+        )).one()
+    assert row[0] == winter
+    assert row[1] == f"season:{winter.year}:한겨울"
+
+
 async def test_generate_now_does_not_append_a_publish_link(client, db_engine, auth_headers, monkeypatch):
     """미니 블로그에 올라가는 글에는 링크를 붙이지 않는다(2026-09-21, 사장님 지시).
     site_payload.publish_url() 자체는 남겨둔다 — 나중에 쓰레드 연동에서 따로 쓸 수 있게."""
     from services import blog_service
 
-    async def fake_generate_one(*, place_name, region, topic_kind, material, used_topics, place_category):
+    async def fake_generate_one(*, place_name, region, topic_kind, material, used_topics, place_category, post_date=None):
         return (f"테스트로 만든 문구입니다. {BODY}", "gemini-test-model")
 
     monkeypatch.setattr(blog_service, "generate_one", fake_generate_one)
@@ -599,7 +639,7 @@ async def test_generate_now_is_noop_for_site_without_domain(client, db_engine, a
     실제로 성공하도록 목킹해 둬야 "그냥 LLM 이 설정 안 돼서 0건"과 구분된다."""
     from services import blog_service
 
-    async def fake_generate_one(*, place_name, region, topic_kind, material, used_topics, place_category):
+    async def fake_generate_one(*, place_name, region, topic_kind, material, used_topics, place_category, post_date=None):
         return ("테스트로 만든 문구입니다. " + BODY, "gemini-test-model")
 
     monkeypatch.setattr(blog_service, "generate_one", fake_generate_one)
@@ -720,7 +760,7 @@ async def test_generation_history_counts_by_batch(client, db_engine, auth_header
     생성했는지" / "어느 모델썼는지 등등" → JSONB 한 칸(generation_meta)에 담는다)."""
     from services import blog_service
 
-    async def fake_generate_one(*, place_name, region, topic_kind, material, used_topics, place_category):
+    async def fake_generate_one(*, place_name, region, topic_kind, material, used_topics, place_category, post_date=None):
         return ("테스트로 만든 문구입니다. " + BODY, "gemini-test-model")
 
     monkeypatch.setattr(blog_service, "generate_one", fake_generate_one)
@@ -819,7 +859,7 @@ async def test_generate_one_fills_a_specific_empty_date(client, db_engine, auth_
     """사장님 지시: "개별적으로 새로 만들수있게 해줘" — 달력에서 빈 날짜 하나만 콕 집어 채운다."""
     from services import blog_service
 
-    async def fake_generate_one(*, place_name, region, topic_kind, material, used_topics, place_category):
+    async def fake_generate_one(*, place_name, region, topic_kind, material, used_topics, place_category, post_date=None):
         return ("테스트로 만든 문구입니다. " + BODY, "gemini-test-model")
 
     monkeypatch.setattr(blog_service, "generate_one", fake_generate_one)
@@ -847,7 +887,7 @@ async def test_generate_one_fails_when_date_already_taken(client, db_engine, aut
     """이미 그 날짜에 글이 있으면(유니크 충돌) 조용히 덮지 않고 실패로 답한다."""
     from services import blog_service
 
-    async def fake_generate_one(*, place_name, region, topic_kind, material, used_topics, place_category):
+    async def fake_generate_one(*, place_name, region, topic_kind, material, used_topics, place_category, post_date=None):
         return ("테스트로 만든 문구입니다. " + BODY, "gemini-test-model")
 
     monkeypatch.setattr(blog_service, "generate_one", fake_generate_one)

solution/backend/tests/test_blog_post.py

@@ -10,12 +10,12 @@
   예전처럼 "GET 은 확인 화면만" 을 기대하는 테스트를 두지 않는다.
 """
 import uuid
-from datetime import timezone
+from datetime import date, timezone
 
 from sqlalchemy import text
 
-from common.enums import JobType, PlaceCategory, PostStatus, PostTopicKind
+from common.enums import JobType, LocalContentType, PlaceCategory, PostStatus, PostTopicKind, SiteStatus
-from services import blog_service
+from services import blog_service, site_payload
 
 BODY = (
     "비가 한 차례 지나간 뒤 마당 돌이 검게 젖었습니다. 이런 날에는 대청마루에 앉아 빗소리만 들어도 "
@@ -51,6 +51,101 @@ def test_filter_drops_same_topic_twice_in_one_batch():
     assert "주제 중복" in dropped[0][1]
 
 
+def test_build_prompt_falls_back_to_stay_wording_for_unmapped_category():
+    """업종별로 분기할 수 있는 구조만 갖춘다 — 숙소 외 업종은 아직 실구현 전이라 숙소 문구로 대체한다."""
+    stay = blog_service.build_prompt(
+        place_name="테스트펜션", region="군산", topic_kind=PostTopicKind.WEATHER.value,
+        material="맑음인 날", used_topics=[], place_category=PlaceCategory.LODGING.value,
+    )
+    restaurant = blog_service.build_prompt(
+        place_name="테스트식당", region="군산", topic_kind=PostTopicKind.WEATHER.value,
+        material="맑음인 날", used_topics=[], place_category=PlaceCategory.RESTAURANT.value,
+    )
+    assert "숙소" in stay
+    assert "숙소" in restaurant  # 음식점 전용 문구가 아직 없어 숙소 문구로 폴백한다
+
+
+def _snapshot_with(*rows):
+    return {"local": {"region_code": "52130", "contents": list(rows)}}
+
+
+def _festival_row(name, start, end):
+    return {"content_type": LocalContentType.FESTIVAL.value, "title": name,
+            "body": {"name": name, "eventstartdate": start, "eventenddate": end, "location": "군산 원도심"}}
+
+
+def test_materials_follow_the_post_date():
+    """게시일에 맞는 소재만 나온다(2026-09-23) — 9월 글에 '한겨울'·'눈'·끝난 축제가 붙지 않는다."""
+    snapshot = _snapshot_with(
+        _festival_row("시간여행축제", "20261002", "20261004"),   # 곧 열린다(9일 뒤)
+        _festival_row("벚꽃축제", "20260401", "20260405"),       # 이미 끝났다
+        _festival_row("겨울빛축제", "20261220", "20261231"),     # 아직 한참 멀다
+        {"content_type": LocalContentType.ATTRACTION.value, "title": "경암동 철길마을",
+         "body": {"name": "경암동 철길마을", "description": "철길 옆 골목"}},
+    )
+
+    out = blog_service.materials(snapshot, date(2026, 9, 23))
+    keys = [key for _kind, key, _material in out]
+
+    assert keys[0] == "festival:2026:시간여행축제"
+    assert "2026.10.02 ~ 2026.10.04" in out[0][2]
+    assert not any("벚꽃" in k or "겨울빛" in k for k in keys)
+    assert "season:2026:초가을" in keys
+    assert not any(k.startswith("season:") and not k.endswith("초가을") for k in keys)
+    assert "nearby:경암동 철길마을" in keys
+    assert "weather:2026-09:비" in keys
+    assert not any(k.endswith(":눈") or k.endswith(":소나기") for k in keys)
+
+
+def test_materials_allow_snow_in_winter_and_reuse_seasons_each_year():
+    winter = [key for _k, key, _m in blog_service.materials(_snapshot_with(), date(2027, 1, 10))]
+
+    assert "weather:2027-01:눈" in winter
+    assert "season:2027:한겨울" in winter
+
+
+def test_build_prompt_tells_the_post_date():
+    prompt = blog_service.build_prompt(
+        place_name="테스트펜션", region="군산", topic_kind=PostTopicKind.SEASON.value,
+        material="초가을", used_topics=[], post_date=date(2026, 9, 23),
+    )
+
+    assert "게시일: 2026년 9월 23일(수) · 초가을" in prompt
+
+
+async def test_generate_one_forwards_place_category_into_the_prompt(monkeypatch):
+    """generate_one 은 place_category 를 build_prompt 로 그대로 넘긴다 — 링크는 여기서
+    붙이지 않는다(호출부의 길이 게이트가 이 반환값에 그대로 걸리기 때문에, 붙이면
+    게이트가 링크까지 세어 정상 문구도 버려진다). 공급자는 provider.active() 를 통해
+    고르므로(Gemini 로 고정하지 않는다) 가짜 공급자로 검증한다."""
+    from services.llm import provider
+    from services.llm.types import LlmResult, Usage
+
+    captured = {}
+
+    class FakeLlm:
+        DEFAULT_MODEL = "fake-model"
+
+        @staticmethod
+        def is_configured():
+            return True
+
+        @staticmethod
+        async def generate(client, model, *, prompt, temperature=0.2, **kwargs):
+            captured["prompt"] = prompt
+            return LlmResult(json=None, text=BODY, usage=Usage())
+
+    monkeypatch.setattr(provider, "active", lambda: FakeLlm)
+
+    result = await blog_service.generate_one(
+        place_name="테스트식당", region="군산", topic_kind=PostTopicKind.WEATHER.value,
+        material="맑음인 날", used_topics=[], place_category=PlaceCategory.RESTAURANT.value,
+    )
+
+    assert result == (BODY, "fake-model")
+    assert "숙소" in captured["prompt"]  # 음식점 전용 문구가 아직 없어 숙소 문구로 폴백한다
+
+
 async def _seed_post(db_engine, owner_id, *, status=PostStatus.SENT, expires_days=14):
     place_id, post_id = uuid.uuid4(), uuid.uuid4()
     token, token_hash, expires = blog_service.issue_token()
@@ -94,6 +189,33 @@ async def test_get_approves_immediately(client, db_engine, owner_id):
     assert res.status_code == 200
     assert "올렸습니다" in res.text
     assert await _status(db_engine, post_id) == PostStatus.APPROVED.value
+    # 발행된 사이트가 없으면(이 테스트) 자동 이동 없이 확인 문구만 보여준다.
+    assert "meta http-equiv=\"refresh\"" not in res.text
+
+
+async def test_approve_page_redirects_to_the_published_blog_section(client, db_engine, owner_id):
+    """승인 확인 화면은 5초 뒤 그 업장의 발행된 사이트(미니 블로그 자리)로 자동 이동한다
+    (2026-09-22, 사장님 지시: "이메일 승인후에 블로그가 올라가는 페이지에 5초 정도
+    이후에 연결되도록"). 자동 이동을 못 믿어도 되게 같은 주소를 링크로도 남긴다."""
+    token, post_id = await _seed_post(db_engine, owner_id)
+    async with db_engine.begin() as conn:
+        place_id = (
+            await conn.execute(text("SELECT place_id FROM place_posts WHERE post_id=:p"), {"p": post_id})
+        ).scalar_one()
+        await conn.execute(
+            text(
+                "INSERT INTO sites (site_id, place_id, domain, status) VALUES (:sid, :pid, :dom, :st)"
+            ),
+            {"sid": uuid.uuid4(), "pid": place_id, "dom": "blog-redirect-test", "st": SiteStatus.PUBLISHED.value},
+        )
+
+    res = await client.get(f"/v1/site/post/approve?t={token}")
+
+    assert res.status_code == 200
+    assert "올렸습니다" in res.text
+    expected_url = site_payload.publish_origin() + "/s/blog-redirect-test#blog"
+    assert f'<meta http-equiv="refresh" content="5;url={expected_url}">' in res.text
+    assert f"href='{expected_url}'" in res.text
 
 
 async def test_approve_link_is_single_use(client, db_engine, owner_id):

solution/backend/tests/test_kakao_webhook.py

@@ -0,0 +1,372 @@
+"""카카오톡 채널 웹훅.
+
+여기서 지키는 것 셋:
+  1. 시크릿 없는 요청은 아무것도 하지 못한다 — 오픈빌더가 서명을 주지 않으므로 이게 유일한 문이다
+  2. 연결되지 않은 발화자는 어떤 사장님도 되지 못한다
+  3. 확인(SEMI)은 **만료되면 안 먹는다** — 한참 뒤의 "네" 한 마디에 묵은 발행이 돌면 안 된다
+"""
+
+import uuid
+from types import SimpleNamespace
+from unittest.mock import AsyncMock
+
+import pytest
+from sqlalchemy import text
+
+from services import kakao_link_service as link_service
+from services.agent import channel, runtime
+
+SECRET = "test-webhook-secret-0123456789"
+PATH = "/v1/agent/kakao/webhook"
+
+
+@pytest.fixture(autouse=True)
+def secret(monkeypatch):
+    monkeypatch.setenv("KAKAO_WEBHOOK_SECRET", SECRET)
+    monkeypatch.setenv("KAKAO_CHANNEL_PUBLIC_ID", "_testCh")
+    monkeypatch.setenv("AGENT_CHAT_ENABLED", "1")
+    monkeypatch.delenv("KAKAO_BOT_ID", raising=False)
+
+
+def body(utterance, speaker="kakao-speaker-1", bot_id="bot-1"):
+    return {
+        "userRequest": {"utterance": utterance, "user": {"id": speaker, "type": "botUserKey"}},
+        "bot": {"id": bot_id, "name": "ADO2"},
+    }
+
+
+def said(res) -> str:
+    return res.json()["template"]["outputs"][0]["simpleText"]["text"]
+
+
+def quick(res) -> list:
+    return [q["label"] for q in res.json()["template"].get("quickReplies", [])]
+
+
+async def owner_with_place(client, auth_headers, name="대화숙소"):
+    h = await auth_headers(f"kakao-{uuid.uuid4().hex[:8]}")
+    res = await client.post("/v1/place", headers=h, json={"name": name, "category": 1})
+    return h, res.json()["place"]["place_id"]
+
+
+async def user_id_of(db_engine, place_id):
+    async with db_engine.begin() as c:
+        return (
+            await c.execute(text("SELECT owner_user_id FROM places WHERE place_id=:p"), {"p": uuid.UUID(place_id)})
+        ).scalar_one()
+
+
+async def link(db_engine, client, auth_headers, speaker, name="대화숙소"):
+    """사장님 하나 + 가게 하나 + 그 사장님에 묶인 카톡 발화자."""
+    h, pid = await owner_with_place(client, auth_headers, name)
+    uid = await user_id_of(db_engine, pid)
+    await link_service.redeem((await link_service.issue_code(uid))["code"], speaker)
+    return h, pid, uid
+
+
+# ── 1. 시크릿이 유일한 문이다 ────────────────────────────────────────────
+
+async def test_시크릿이_없으면_존재를_알리지_않는다(client, monkeypatch):
+    """★ 401 이 아니라 404 다. 401 은 '여기 뭔가 있다' 를 알려 준다."""
+    monkeypatch.setenv("KAKAO_WEBHOOK_SECRET", "")
+    assert (await client.post(PATH, json=body("안녕"))).status_code == 404
+
+
+async def test_틀린_시크릿도_404(client):
+    res = await client.post(PATH, headers={"X-Agent-Secret": "wrong-secret"}, json=body("안녕"))
+    assert res.status_code == 404
+    res = await client.post(PATH, json=body("안녕"))  # 헤더 없음
+    assert res.status_code == 404
+
+
+async def test_경로_시크릿도_받는다(client, db_engine):
+    res = await client.post(f"{PATH}/{SECRET}", json=body("안녕"))
+    assert res.status_code == 200
+    res = await client.post(f"{PATH}/wrong-secret", json=body("안녕"))
+    assert res.status_code == 404
+
+
+async def test_다른_봇의_요청은_거절한다(client, monkeypatch, db_engine):
+    monkeypatch.setenv("KAKAO_BOT_ID", "bot-1")
+    ok = await client.post(PATH, headers={"X-Agent-Secret": SECRET}, json=body("안녕", bot_id="bot-1"))
+    assert ok.status_code == 200
+    bad = await client.post(PATH, headers={"X-Agent-Secret": SECRET}, json=body("안녕", bot_id="남의봇"))
+    assert bad.status_code == 404
+
+
+# ── 2. 연결되지 않은 발화자 ──────────────────────────────────────────────
+
+async def test_연결_전에는_어떤_도구도_돌지_않는다(client, monkeypatch, db_engine):
+    called = AsyncMock()
+    monkeypatch.setattr(runtime, "chat", called)
+    res = await client.post(PATH, headers={"X-Agent-Secret": SECRET}, json=body("체크인 3시로 바꿔줘", "모르는-키"))
+    assert res.status_code == 200
+    assert "연결" in said(res)
+    called.assert_not_awaited()
+
+
+async def test_발화자_id_를_위조해도_남의_가게에_닿지_않는다(client, auth_headers, db_engine, monkeypatch):
+    """★ 신원 연결이 없으면 카톡 진입점만 소유자 범위 밖에 놓인다 — 그걸 막는 자리다."""
+    await link(db_engine, client, auth_headers, "진짜-사장님-키")
+    called = AsyncMock()
+    monkeypatch.setattr(runtime, "chat", called)
+    res = await client.post(PATH, headers={"X-Agent-Secret": SECRET}, json=body("정보 보여줘", "위조한-키"))
+    assert "연결" in said(res)
+    called.assert_not_awaited()
+
+
+async def test_코드를_보내면_연결된다(client, auth_headers, db_engine):
+    h, pid = await owner_with_place(client, auth_headers)
+    uid = await user_id_of(db_engine, pid)
+    code = (await link_service.issue_code(uid))["code"]
+
+    res = await client.post(PATH, headers={"X-Agent-Secret": SECRET}, json=body(code, "새-발화자"))
+    assert "연결됐습니다" in said(res)
+    assert await link_service.resolve("새-발화자") == uid
+
+
+async def test_틀린_코드는_이유를_구분해_말하지_않는다(client, db_engine):
+    res = await client.post(PATH, headers={"X-Agent-Secret": SECRET}, json=body("ZZZZZZ", "새-발화자2"))
+    assert "코드가 맞지 않거나" in said(res)
+
+
+# ── 3. 확인은 만료되면 안 먹는다 ─────────────────────────────────────────
+
+async def test_발행은_묻고_바로가기를_준다(client, auth_headers, db_engine, monkeypatch):
+    speaker = "확인-테스트-키"
+    await link(db_engine, client, auth_headers, speaker)
+    # 테스트는 실제 모델을 부르지 않는다 — 런타임만 열고 선택 결과를 대신 준다.
+    monkeypatch.setattr(runtime, "is_configured", lambda: True)
+    monkeypatch.setattr(runtime, "_choose", AsyncMock(return_value={"tool": "publish", "args": {}, "message": ""}))
+
+    res = await client.post(PATH, headers={"X-Agent-Secret": SECRET}, json=body("발행해줘", speaker))
+    assert channel.CONFIRM_LABEL in quick(res)
+    async with db_engine.begin() as c:
+        pending = (
+            await c.execute(text("SELECT pending_tool FROM owner_kakao_links WHERE channel_user_key=:k"), {"k": speaker})
+        ).scalar_one()
+    assert pending == "publish"
+
+
+async def test_만료된_확인에_네_라고_해도_실행되지_않는다(client, auth_headers, db_engine, monkeypatch):
+    """★ 이게 없으면 한참 뒤의 '네' 한 마디에 **묵은 발행**이 돈다."""
+    speaker = "만료-테스트-키"
+    await link(db_engine, client, auth_headers, speaker)
+    async with db_engine.begin() as c:
+        await c.execute(
+            text("""UPDATE owner_kakao_links
+                       SET pending_tool='publish', pending_args='{}'::jsonb,
+                           pending_expires_at = now() - interval '1 minute'
+                     WHERE channel_user_key=:k"""),
+            {"k": speaker},
+        )
+    ran = AsyncMock()
+    monkeypatch.setattr(runtime, "chat", ran)
+    await client.post(PATH, headers={"X-Agent-Secret": SECRET}, json=body("네", speaker))
+    # 확인으로 실행된 적이 없다(다른 말로 취급돼 일반 경로로 갔을 수는 있다).
+    for call in ran.await_args_list:
+        assert call.kwargs.get("confirm") is None
+
+
+async def test_바로가기_라벨은_예로_읽히는_말에_들어_있다():
+    """★ 라벨과 _YES 가 어긋나면 **눌러도 안 먹는다** — 사장님은 버튼이 고장난 줄 안다."""
+    assert channel.CONFIRM_LABEL in channel._YES
+    assert channel.PUBLISH_LABEL in channel._YES
+    assert channel.DECLINE_LABEL in channel._NO
+
+
+# ── 가게 고르기 ──────────────────────────────────────────────────────────
+
+async def test_가게가_여럿이면_추측하지_않고_되묻는다(client, auth_headers, db_engine, monkeypatch):
+    """★ 임의로 첫 가게를 고르면 사장님은 엉뚱한 가게를 고쳐 놓고도 모른다."""
+    speaker = "다가게-키"
+    h, _pid, _uid = await link(db_engine, client, auth_headers, speaker, "첫째가게")
+    await client.post("/v1/place", headers=h, json={"name": "둘째가게", "category": 1})
+
+    called = AsyncMock()
+    monkeypatch.setattr(runtime, "chat", called)
+    res = await client.post(PATH, headers={"X-Agent-Secret": SECRET}, json=body("정보 보여줘", speaker))
+    assert "어느 가게" in said(res)
+    assert set(quick(res)) == {"첫째가게", "둘째가게"}
+    called.assert_not_awaited()
+
+    # 고르면 기억한다.
+    picked = await client.post(PATH, headers={"X-Agent-Secret": SECRET}, json=body("둘째가게", speaker))
+    assert "둘째가게" in said(picked)
+    async with db_engine.begin() as c:
+        current = (
+            await c.execute(
+                text("SELECT current_place_id FROM owner_kakao_links WHERE channel_user_key=:k"), {"k": speaker}
+            )
+        ).scalar_one()
+    assert current is not None
+
+
+# ── 5초 벽 · 실패 ────────────────────────────────────────────────────────
+
+async def test_느리면_침묵_대신_안내로_끊는다(client, monkeypatch, db_engine):
+    """넘기면 카카오가 연결을 끊는다 — 사장님에게는 말없이 실패하는 봇이 된다."""
+    import router.v1.agent.kakao_bot as bot
+
+    async def slow(*_a, **_kw):
+        import asyncio
+
+        await asyncio.sleep(1)
+
+    monkeypatch.setattr(bot, "DEADLINE_SEC", 0.01)
+    monkeypatch.setattr(bot.channel, "handle", slow)
+    res = await client.post(PATH, headers={"X-Agent-Secret": SECRET}, json=body("안녕"))
+    assert res.status_code == 200
+    assert "잠시 뒤" in said(res)
+
+
+async def test_내부_오류도_200_으로_답한다(client, monkeypatch, db_engine):
+    """메신저에서는 500 도 침묵으로 보인다 — 무슨 일이 있었는지 한 줄은 말해야 한다."""
+    import router.v1.agent.kakao_bot as bot
+
+    monkeypatch.setattr(bot.channel, "handle", AsyncMock(side_effect=RuntimeError("어딘가 터짐")))
+    res = await client.post(PATH, headers={"X-Agent-Secret": SECRET}, json=body("안녕"))
+    assert res.status_code == 200
+    assert "어딘가 터짐" not in res.text
+
+
+async def test_발화자가_없으면_거기서_끝낸다(client, db_engine):
+    res = await client.post(
+        PATH, headers={"X-Agent-Secret": SECRET}, json={"userRequest": {"utterance": "안녕", "user": {}}}
+    )
+    assert res.status_code == 200
+    assert "확인하지 못했" in said(res)
+
+
+# ── 응답 형식 ────────────────────────────────────────────────────────────
+
+def test_카카오_형식은_이_파일_밖으로_나가지_않는다():
+    """★ 서비스 계층에 새면 다른 채널을 붙일 때 전부 걷어내야 한다."""
+    import ast
+    import inspect
+
+    # ★ 주석·docstring 에 이름이 나오는 것은 '샌' 것이 아니다 — 실제 코드만 본다.
+    tree = ast.parse(inspect.getsource(channel))
+    for node in ast.walk(tree):
+        if isinstance(node, ast.Expr) and isinstance(node.value, ast.Constant) and isinstance(node.value.value, str):
+            node.value.value = ""  # docstring 비우기
+    dumped = ast.dump(tree)
+    for token in ("simpleText", "quickReplies", "userRequest", "2.0"):
+        assert token not in dumped, token
+
+
+def test_바로가기는_열_개를_넘기지_않는다():
+    import router.v1.agent.kakao_bot as bot
+
+    out = bot._reply("안녕", [f"라벨{i}" for i in range(20)])
+    assert len(out["template"]["quickReplies"]) == 10
+    assert out["version"] == "2.0"
+    assert out["template"]["outputs"][0]["simpleText"]["text"] == "안녕"
+
+
+# ── 사이트 목록 · 고르기 ─────────────────────────────────────────────────
+
+async def test_연결되자마자_홈페이지_목록을_알려준다(client, auth_headers, db_engine):
+    """★ 연결만 알리고 끝내면 사장님은 **어느 홈페이지를 다루는 대화인지** 모른 채 말을 건다."""
+    h, pid = await owner_with_place(client, auth_headers, "첫째가게")
+    await client.post("/v1/place", headers=h, json={"name": "둘째가게", "category": 1})
+    uid = await user_id_of(db_engine, pid)
+    code = (await link_service.issue_code(uid))["code"]
+
+    res = await client.post(PATH, headers={"X-Agent-Secret": SECRET}, json=body(code, "목록-발화자"))
+    text_out = said(res)
+    assert "연결됐습니다" in text_out
+    assert "첫째가게" in text_out and "둘째가게" in text_out
+    # 바로 고를 수 있어야 한다 — 이름을 외워 치게 하지 않는다.
+    assert set(quick(res)) == {"첫째가게", "둘째가게"}
+
+
+async def test_가게가_하나면_그_이름을_말해_준다(client, auth_headers, db_engine):
+    h, pid = await owner_with_place(client, auth_headers, "혼자가게")
+    uid = await user_id_of(db_engine, pid)
+    code = (await link_service.issue_code(uid))["code"]
+    res = await client.post(PATH, headers={"X-Agent-Secret": SECRET}, json=body(code, "하나-발화자"))
+    assert "혼자가게" in said(res)
+    assert "발행" in said(res)  # 발행 여부를 같이 말한다
+
+
+async def test_목록이라고_하면_언제든_다시_보여주고_가게를_바꿀_수_있다(client, auth_headers, db_engine, monkeypatch):
+    """★ 대화가 막혔을 때 사장님이 처음 찾는 길이다. LLM 을 부르지 않는다."""
+    speaker = "전환-발화자"
+    h, _pid, _uid = await link(db_engine, client, auth_headers, speaker, "첫째가게")
+    await client.post("/v1/place", headers=h, json={"name": "둘째가게", "category": 1})
+
+    called = AsyncMock()
+    monkeypatch.setattr(runtime, "chat", called)
+
+    # 먼저 한 곳을 고른다.
+    await client.post(PATH, headers={"X-Agent-Secret": SECRET}, json=body("첫째가게", speaker))
+    # 목록으로 돌아온다.
+    res = await client.post(PATH, headers={"X-Agent-Secret": SECRET}, json=body("가게 바꿔줘", speaker))
+    assert "첫째가게" in said(res) and "둘째가게" in said(res)
+    assert set(quick(res)) == {"첫째가게", "둘째가게"}
+    called.assert_not_awaited()  # 목록 보기에 모델을 부르지 않는다
+
+    # 다른 곳으로 바꾼다.
+    picked = await client.post(PATH, headers={"X-Agent-Secret": SECRET}, json=body("둘째가게", speaker))
+    assert "둘째가게" in said(picked)
+
+
+async def test_목록은_발행_여부를_같이_말한다(client, auth_headers, db_engine):
+    """★ 안 그러면 사장님은 고친 것이 손님에게 보이는 줄 안다."""
+    speaker = "발행표시-발화자"
+    await link(db_engine, client, auth_headers, speaker, "미발행가게")
+    res = await client.post(PATH, headers={"X-Agent-Secret": SECRET}, json=body("목록", speaker))
+    assert "아직 발행 전" in said(res)
+
+
+# ── 콜백 (5초 벽 넘기) ───────────────────────────────────────────────────
+
+async def test_콜백이_켜져_있으면_즉답하고_뒤에서_마저_만든다(client, auth_headers, db_engine, monkeypatch):
+    """★ 오픈빌더 스킬 타임아웃은 5초다. 콜백을 쓰면 '잠시만 기다려 주세요' 를 먼저 주고
+    답이 완성되면 그 주소로 따로 보낸다 — 느린 답이 침묵이 되지 않는다."""
+    import router.v1.agent.kakao_bot as bot
+
+    sent = {}
+
+    async def fake_push(callback_url, utterance, speaker):
+        sent["url"] = callback_url
+        sent["payload"] = await bot._answer(utterance, speaker, 5.0)
+
+    monkeypatch.setattr(bot, "_push", fake_push)
+
+    payload = body("안녕", "콜백-발화자")
+    payload["userRequest"]["callbackUrl"] = "https://callback.example/one-shot"
+    res = await client.post(PATH, headers={"X-Agent-Secret": SECRET}, json=payload)
+
+    # 즉답은 useCallback 이다 — template 을 싣지 않는다.
+    assert res.json()["useCallback"] is True
+    assert res.json()["data"]["text"]
+    assert "template" not in res.json()
+
+    # 실제 답은 콜백으로 간다.
+    assert sent["url"] == "https://callback.example/one-shot"
+    assert "연결" in sent["payload"]["template"]["outputs"][0]["simpleText"]["text"]
+
+
+async def test_콜백이_없으면_예전처럼_동기로_답한다(client, db_engine):
+    res = await client.post(PATH, headers={"X-Agent-Secret": SECRET}, json=body("안녕", "동기-발화자"))
+    assert "useCallback" not in res.json()
+    assert res.json()["template"]["outputs"]
+
+
+async def test_콜백_전송이_실패해도_터지지_않는다(monkeypatch):
+    """★ 주소는 1분 · 1회다. 재시도하지 않는다 — 두 번째 POST 는 어차피 거절된다."""
+    import router.v1.agent.kakao_bot as bot
+
+    monkeypatch.setattr(bot, "_answer", AsyncMock(return_value=bot._reply("답")))
+
+    class Boom:
+        async def __aenter__(self):
+            raise RuntimeError("네트워크 끊김")
+
+        async def __aexit__(self, *_):
+            return False
+
+    monkeypatch.setattr(bot.httpx, "AsyncClient", lambda **_kw: Boom())
+    await bot._push("https://callback.example/x", "안녕", "누구")  # 예외가 새 나오지 않는다