[feat] solution/backend,docs: 미니블로그 승인 → 쓰레드 자동 게재, 쓰레드 초안 생성 OpenAI 전환

사장님 지시: "쓰레드에 연동되어 있으면 같이 업로드 되는 기능". 미니블로그의 두
승인 경로(이메일 GET 토큰, 로그인 "바로 발행")를 공통 메서드로 묶고, 그 끝에서
쓰레드 연동을 시도한다. 미니블로그 승인 자체가 발화 동의로 취급되므로 쓰레드
쪽 별도 승인은 묻지 않는다(DECISIONS 7-1-2 개정, 문구를 그대로 재사용하는
경우에 한정). 겸사겸사 쓰레드 초안 생성(generate_social_post)이 LLM_PROVIDER
를 안 타고 Gemini 를 직접 호출하던 것도 다른 생성 함수와 같은 추상화로 맞췄다.

- blog_jobs._published_places: sites.domain IS NOT NULL 조건 추가(쓰레드 기준과 통일)
- social_service.publish_reused_text: 연동 없음/게시 비활성/domain 미확정이면 스킵,
  정상이면 APPROVED 삽입 + run_post(job_type=9) enqueue — 새 게시 로직은 안 만든다
- post_service: decide/approve_by_owner → _approve_and_publish 로 공통화,
  _try_social_share 는 실패를 전부 삼켜 미니블로그 승인을 막지 않는다
- gemini_text.generate_social_post: services.llm.provider.active() 로 전환,
  Gemini 전용 import 제거
- DECISIONS.md 7-1-2, MINI_BLOG.md 5·8절, SOCIAL.md 갱신

test_blog_owner.py·test_social.py 다수 추가/수정, 관련 스위트 전체 PASS

docs/DECISIONS.md

@@ -346,6 +346,13 @@ JWT 를 쓰지 않는 이유는 JWT 가 서명만 검증하고 **소비를 못
 **사람이 누르기 전에** 그 URL 을 연다. GET 승인이면 사장님이 안 눌렀는데 글이 올라가고 로그에는
 "승인됨" 으로 남는다 — 눈으로 원인을 못 찾는 종류다.
 
+**2026-09-21 개정 — 미니블로그 문구 재사용은 예외.** 미니블로그 승인(이메일 GET 토큰 또는
+로그인 "바로 발행")은 "이 문구를 공개해도 좋다"는 사장님의 명시적 의사표시이고, 같은 문구를
+같은 시점에 다른 채널(쓰레드)에도 내보내는 것뿐이므로 별도 승인은 중복 확인이다. 이 예외는
+**미니블로그 문구를 그대로 재사용하는 경우에 한정**한다 — `social_service.publish_reused_text`
+가 `decided_via='mini_blog'`로 곧장 `APPROVED` 처리한다. 쓰레드 전용으로 새로 짓거나 내용을
+바꾸는 경로(`create_draft`/`request_approval`)는 위 CAS 승인을 그대로 거친다.
+
 ★ **기존 액세스 토큰을 승인 링크에 얹지 않는다.** 지금 JWT 는 `sub` 에 `UserInfo` 통짜(role 포함)를
 넣는다 — 그게 링크에 실리면 카톡 전달 한 번이 **빌더 전체 권한 양도**다.
 

docs/MINI_BLOG.md

@@ -120,11 +120,24 @@
 
 ## 5. 승인·수정
 
-- **승인**: GET 이 로그인 없이 즉시 승인한다 → `status = APPROVED` → BUILD 잡 큐. 만료·
+- **게재**: 이메일의 승인 링크(로그인 없음, 누르면 즉시 승인) 또는 빌더 앱에 로그인해
-  재사용은 "처리할 수 없는 링크입니다" 안내로 끝낸다(오류 화면을 주지 않는다)
+  "바로 발행" 버튼을 눌러도 승인된다(2026-09-21, 사장님 지시: "이메일 승인으로도 발행
-- **수정**: 빌더 앱 편집 모달에서 저장 → `is_publishable_body` 재검사 → 통과 시 본문 갱신 +
+  가능하고 바로발행버튼으로도 발행 가능하도록") — 두 경로 다 열려 있다(`post_service.
-  그대로 APPROVED 전환 + BUILD 잡. 실패하면 사유를 보여주고 다시 고치게 한다(사이트에는
+  PostService._approve_and_publish`). PUT(수정)은 저장만 하고 자동으로 승인하지 않는다.
-  안 올라간다)
+- **승인**: 이메일 GET 은 로그인 없이 즉시 승인, "바로 발행" 은 로그인 세션이 신원 →
+  둘 다 `status = APPROVED` → BUILD 잡 큐. 이메일 링크의 만료·재사용은 "처리할 수 없는
+  링크입니다" 안내로 끝낸다(오류 화면을 주지 않는다)
+- **쓰레드 연동**: 승인되는 순간(경로 무관) 그 업장이 쓰레드에 연결돼 있으면 같은 문구에
+  발행 링크를 붙여 쓰레드에도 즉시 게시한다(2026-09-21) — 별도 승인 없음(`docs/DECISIONS.md`
+  7-1-2 개정, `docs/SOCIAL.md`). 연동 안 돼 있거나 `SOCIAL_POSTING_ENABLED=0`이거나 사이트
+  domain이 미확정이면 조용히 건너뛴다. 실패해도 미니블로그 승인 자체는 막지 않는다
+  (`post_service.PostService._try_social_share`)
+- **수정**: 빌더 앱 편집 모달에서 저장 → `is_publishable_body` 재검사 → 통과 시 본문만
+  갱신한다. **승인 전환은 하지 않는다** — 실패하면 사유를 보여주고 다시 고치게 한다,
+  통과해도 두 승인 경로 중 하나를 눌러야 사이트에 반영된다
+- **알림 이메일**: 승인 메일 수신자는 `places.notify_email`(비면 `users.email`) —
+  계정 로그인 이메일과 분리해서 업장별로 다른 담당자에게 보낼 수 있다(빌더 앱 미니블로그
+  관리 화면에서 수정, `PATCH /v1/place/{place_id}`)
 - ★ BUILD 잡 payload 에는 반드시 `owner_user_id` 가 있어야 한다(`build_service.run_build`
   가 `payload["owner_user_id"]` 를 무조건 읽는다) — 토큰/day-pass 흐름은 일반 로그인
   세션과 달라 `post_service.PostService._enqueue_build` 가 `place_id` 로 직접 조회해
@@ -213,7 +226,8 @@
   날짜는 LLM 을 부르지 않고 건너뛰고, 구간 안 소재가 떨어지면 그 자리에서 멈춘다 — 응답에
   `requested`(구간 일수)·`created`(실제로 채운 일수)를 같이 줘서 "N일 중 M일만 채웠습니다"로
   보여준다. 발행 전 사업장은 애초에 생성 스윕 대상이 아니라(`_published_places`) 여기서도
-  0건이다
+  0건이다. domain 이 아직 확정되지 않은(임시 주소) 사이트도 마찬가지다(2026-09-21 —
+  쓰레드 연동 요구사항과 맞췄다, `docs/SOCIAL.md` 7-1-1과 동일 기준)
 
 ## 8. 진행 (2026-09-17)
 
@@ -230,6 +244,7 @@
 | 개별 생성(빈 날짜 하나) | `POST .../post/generate-one` · `blog_jobs.generate_one_for_date` | 완료 |
 | payload + 화면 | `site_payload.posts[]` · `site/src/sections/BlogSection.tsx` | 완료 |
 | 재발행 연결 | `build_service` → `mark_published`, `owner_user_id` 버그 수정 | 완료 |
+| 쓰레드 자동 게재 | `services/social_service.publish_reused_text` · `post_service._try_social_share` | 완료 |
 
 남은 것: 운영 ACS 에 발신 도메인 등록(지금은 negodata 리소스를 빌려 쓴다),
 그리고 6절의 발행일 갱신 합의.

docs/SOCIAL.md

@@ -14,6 +14,13 @@ Meta 개발자 문서 일부는 조사 시 429를 반환했다. 실제 앱 권
 문자열은 NFC로 정규화하고 초과하면 최대 3번 다시 요청한다. 잘라서 게시하지 않는다.
 같은 사업장·발행 버전은 성공 이후에도 원고 1건만 유지한다. 초안 생성 실패만 같은 행으로 재시도한다.
 
+미니블로그 승인(이메일 링크 또는 빌더 앱 "바로 발행")도 계정이 연결돼 있으면 같은 문구를
+그대로 쓰레드에 낸다(`decided_via='mini_blog'`) — 이 경로는 승인 요청·알림톡을 거치지 않고
+바로 `APPROVED`로 들어간다. 미니블로그 승인 자체가 발화 동의로 취급되기 때문이다(2026-09-21,
+DECISIONS 7-1-2 개정 — 문구를 그대로 재사용하는 경우에 한정). 쓰레드 전용으로 새로 짓거나
+내용을 바꾸는 경로(위 "발행 모달 → 소개글 쓰기")는 여전히 계정 연결 → 승인 요청 → 명시적
+승인을 그대로 거친다.
+
 - 계약 없이: 초안 작성, 복사, 화면에서 내용 확인/거절, 만료 후 재요청.
 - 실제 연결 이후: Threads 계정 연결 → 게재 승인 요청 → 화면 또는 알림톡 확인 → 명시적 POST 승인 → 게시.
 - 계정 미연결 상태의 내용 확인은 게시를 예약하지 않는다. 연결한 뒤 계정을 보여주고 다시 승인받는다.

solution/backend/services/blog_jobs.py

@@ -13,6 +13,7 @@ from datetime import date, datetime, timedelta, timezone
 
 from sqlalchemy import select
 
+from config import social_config
 from common.database.db_session_manager import DB_SESSION_MNG
 from common.database.model.models import place_posts, places, sites, users
 from common.enums import DBWRType, PostStatus, SiteStatus
@@ -49,6 +50,7 @@ async def _published_places() -> list:
                 places.deleted == False,  # noqa: E712
                 sites.deleted == False,  # noqa: E712
                 sites.status == SiteStatus.PUBLISHED.value,
+                sites.domain.isnot(None),
             )
         )
 
@@ -92,7 +94,7 @@ async def _generate_for_place(place) -> int:
             continue
         generated = await blog_service.generate_one(
             place_name=place.name, region=region, topic_kind=kind,
-            material=material, used_topics=sorted(used_set),
+            material=material, used_topics=sorted(used_set), place_category=place.category,
         )
         if not generated:
             continue
@@ -171,7 +173,7 @@ async def generate_range(place_id: str, start_date: date, end_date: date) -> dic
             continue
         generated = await blog_service.generate_one(
             place_name=place.name, region=region, topic_kind=kind,
-            material=material, used_topics=sorted(used_set),
+            material=material, used_topics=sorted(used_set), place_category=place.category,
         )
         if not generated:
             continue
@@ -219,7 +221,7 @@ async def generate_one_for_date(place_id: str, target_date: date) -> dict | None
             continue
         generated = await blog_service.generate_one(
             place_name=place.name, region=region, topic_kind=kind,
-            material=material, used_topics=sorted(used_set),
+            material=material, used_topics=sorted(used_set), place_category=place.category,
         )
         if not generated:
             continue
@@ -261,6 +263,41 @@ def _mail_body(*, place_name: str, post, user, origin: str, approve_token: str)
     )
 
 
+def _notify_address(place, user) -> str:
+    # notify_email 이 있으면 그 업장 전용 수신자다 — 없으면 계정 이메일(users.email)로 대체한다
+    # (사장님 한 명이 사이트를 여러 개 가질 수 있어 계정 이메일 하나로는 업장별 수신자를 못 나눈다).
+    return place.notify_email or user.email
+
+
+def _app_origin() -> str:
+    """메일의 승인·수정 링크가 향할 곳 — 빌더 앱(과 그 앞의 API)이 사는 오리진.
+
+    ★ site_payload.publish_origin() 을 쓰면 안 된다 — 그건 발행된 고객 사이트(/s/<slug>)
+      전용이다. 로컬에선 그게 solution-site 정적 서버(포트 80)라, 메일의 "수정하려면"
+      링크(/blog?...)가 거기로 가서 404 났다(2026-09-21 실측). SNS 알림(notify_service.py)이
+      이미 같은 목적으로 쓰는 SOCIAL_APP_ORIGIN 을 그대로 재사용한다 — 설정을 두 벌 안 둔다.
+      비어 있으면(로컬에서 안 채웠으면) publish_origin() 으로 폴백해 링크가 아예 상대경로로
+      깨지는 것보다는 낫게 한다."""
+    return social_config.get("SOCIAL_APP_ORIGIN") or site_payload.publish_origin()
+
+
+async def _send_one(place, user, post) -> bool:
+    """토큰 발급 → 메일 본문 조립 → 발송 → 성공하면 SENT 로 표시. 실패하면 DB 를 안 건드린다."""
+    token, token_hash, expires = blog_service.issue_token()
+    body = _mail_body(
+        place_name=place.name, post=post, user=user,
+        origin=_app_origin(), approve_token=token,
+    )
+    ok = mail_service.send(to=_notify_address(place, user), subject=f"[{place.name}] 이번 글 올릴까요?", text=body)
+    if not ok:
+        return False
+    await DB_SESSION_MNG.execute_lambda_run(
+        [place_posts.DBType()],
+        [lambda s, pid=post.post_id, h=token_hash, e=expires: _crud.mark_sent(s, pid, h, e)],
+    )
+    return True
+
+
 async def send_reviewed() -> int:
     """검수를 통과한 글을 사장님에게 한 통씩 보낸다. 보낸 수를 돌려준다."""
     if not mail_service.is_configured():
@@ -273,22 +310,45 @@ async def send_reviewed() -> int:
     if not rows:
         return 0
 
-    targets = {str(place.place_id): (place.name, user) for place, user in await _published_places()}
+    places_by_id = {str(place.place_id): (place, user) for place, user in await _published_places()}
-    origin = site_payload.publish_origin()
     sent = 0
     for post in rows:
-        target = targets.get(str(post.place_id))
+        target = places_by_id.get(str(post.place_id))
-        if not target or not mail_service.is_valid_address(target[1].email or ""):
+        if not target:
             continue
-        place_name, user = target
+        place, user = target
-        token, token_hash, expires = blog_service.issue_token()
+        if not mail_service.is_valid_address(_notify_address(place, user) or ""):
-        body = _mail_body(place_name=place_name, post=post, user=user, origin=origin, approve_token=token)
-        ok = mail_service.send(to=user.email, subject=f"[{place_name}] 이번 글 올릴까요?", text=body)
-        if not ok:
             continue
-        await DB_SESSION_MNG.execute_lambda_run(
+        if await _send_one(place, user, post):
-            [place_posts.DBType()],
-            [lambda s, pid=post.post_id, h=token_hash, e=expires: _crud.mark_sent(s, pid, h, e)],
-        )
             sent += 1
     return sent
+
+
+async def send_now_for_place(place_id: str) -> dict:
+    """사장님이 빌더 화면에서 누르는 즉시 발송 — 아침 9시 스윕을 기다리지 않고 이 업장의
+    오늘 몫을 지금 보낸다(2026-09-21, 사장님 요청: "지금 바로 발송할 수 있도록").
+    '하루 한 통' 원칙은 그대로다 — 이미 오늘 보냈으면(REVIEWED 가 아니면) 보낼 게 없다."""
+    if not mail_service.is_configured():
+        return {"sent": False, "reason": "MAIL_NOT_CONFIGURED"}
+
+    place = user = None
+    for p, u in await _published_places():
+        if str(p.place_id) == str(place_id):
+            place, user = p, u
+            break
+    if place is None:
+        return {"sent": False, "reason": "NOTHING_DUE"}
+
+    post = await DB_SESSION_MNG.execute_lambda(
+        place_posts.DBType(), DBWRType.DB_READ.value,
+        lambda s: _crud.next_due_for_mail(s, place_id, PostStatus.REVIEWED.value, _today_kst()),
+    )
+    if post is None:
+        return {"sent": False, "reason": "NOTHING_DUE"}
+
+    if not mail_service.is_valid_address(_notify_address(place, user) or ""):
+        return {"sent": False, "reason": "NO_VALID_EMAIL"}
+
+    if not await _send_one(place, user, post):
+        return {"sent": False, "reason": "SEND_FAILED"}
+    return {"sent": True, "reason": None}

solution/backend/services/external/gemini_text.py

@@ -25,10 +25,6 @@ from services.llm.errors import LlmError
 from services.llm.errors import LlmInvalidOutput as GeminiInvalidOutput
 from services.llm.errors import LlmNotConfigured as GeminiNotConfigured
 from services.prompts.copy import RESPONSE_SCHEMA, build_prompt
-# generate_social_post 는 공급자 선택(provider.active())을 타지 않고 Gemini 를 직접 부른다 —
-# 구조화 출력 재시도 루프(피드백 프롬프트로 다시 묻기)가 Gemini REST 응답 모양에 맞춰져 있다.
-# 모듈 최상단에서 이름으로 들여와야 테스트가 gemini_text.call 을 monkeypatch 할 수 있다.
-from services.llm.gemini import DEFAULT_MODEL as DEFAULT_TEXT_MODEL, call, extract_text
 
 
 def is_configured() -> bool:
@@ -325,15 +321,22 @@ async def generate_song(
 
 
 async def generate_social_post(place_name, facts, link_url, provider=2, *, client=None):
-    """실제 게시 문자열을 검증한다. 초과·근거 실패 시 다시 받고 문장을 자르지 않는다."""
+    """실제 게시 문자열을 검증한다. 초과·근거 실패 시 다시 받고 문장을 자르지 않는다.
+
+    ★ 2026-09-21: 다른 생성 함수(generate_copy 등)와 같은 이유로 공급자 선택을 탄다
+      (`LLM_PROVIDER`, 기본 openai) — Gemini 고정을 없앴다. 인자 이름 `provider` 는
+      SNS 플랫폼(쓰레드=2)을 가리키는 기존 값이라, LLM 공급자 모듈은 `llm_provider` 로
+      따로 들여와 이름이 겹치지 않게 한다."""
     from services.prompts import social
     from services.external.social import adapter, weighted_length, URL
+    from services.llm import provider as llm_provider
     import unicodedata
 
     if not facts:
         raise GeminiInvalidOutput('NO_GROUNDED_FACTS')
-    if not is_configured():
+    llm = llm_provider.active()
-        raise GeminiNotConfigured('GEMINI_NOT_CONFIGURED')
+    if not llm.is_configured():
+        raise GeminiNotConfigured(f"{llm.__name__.rsplit('.', 1)[-1].upper()}_API_KEY 가 설정되지 않았다")
     limit = adapter(provider).weighted_limit()
     allowed = {f.key for f in facts}
     feedback = ''
@@ -341,14 +344,14 @@ async def generate_social_post(place_name, facts, link_url, provider=2, *, clien
     client = client or httpx.AsyncClient(timeout=45)
     try:
         for _ in range(3):
-            request = {'contents': [{'role': 'user', 'parts': [{'text': social.build_prompt(
+            prompt = social.build_prompt(
-                place_name, facts, limit - weighted_length('\n\n' + link_url, provider), feedback)}]}],
+                place_name, facts, limit - weighted_length('\n\n' + link_url, provider), feedback)
-                'generationConfig': {'responseMimeType': 'application/json',
+            result = await llm.generate(
-                                     'responseSchema': social.RESPONSE_SCHEMA, 'temperature': 0.2,
+                client, llm.DEFAULT_MODEL, prompt=prompt,
-                                     'maxOutputTokens': 1024}}
+                response_schema=social.RESPONSE_SCHEMA, temperature=0.2, max_retries=0,
-            result = await call(client, DEFAULT_TEXT_MODEL, request, max_retries=0)
+            )
             try:
-                parsed = json.loads(extract_text(result))
+                parsed = result.json if result.json is not None else json.loads(result.text)
                 body = unicodedata.normalize('NFC', parsed['body'].strip())
                 keys = parsed['fact_keys']
                 text = body + '\n\n' + link_url

solution/backend/services/post_service.py

@@ -23,7 +23,7 @@ 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
+from services import blog_jobs, blog_service, social_service
 from services.job_service import enqueue_job
 
 EXPIRED = "처리할 수 없는 링크입니다."
@@ -75,10 +75,7 @@ class PostService:
             )
             return {"success": True, "message": SKIPPED}
 
-        await DB_SESSION_MNG.execute_lambda_run(
+        await self._approve_and_publish(post_id, place_id)
-            [place_posts.DBType()], [lambda s: self.crud.approve(s, post_id)],
-        )
-        await self._enqueue_build(post_id, place_id)
         return {"success": True, "message": APPROVED}
 
     async def _load_place(self, user_info: UserInfo, place_id: str):
@@ -191,7 +188,10 @@ class PostService:
     async def edit_by_owner(
         self, user_info: UserInfo, place_id: str, post_id: str, body: str
     ) -> Res_WebPacketProtocol:
-        """로그인 세션으로 직접 고치기 — 메일이 아직 안 나간 REVIEWED 글도 여기서 바로 승인된다."""
+        """로그인 세션으로 직접 고치기 — 저장만 한다. ★ 승인은 여기서 하지 않는다(2026-09-21,
+        사장님 지시: "승인되야 올라가도록 해야 한다") — 저장 후에는 이메일 승인 링크
+        (router/v1/site/post.py approve_page → decide) 또는 바로 아래 approve_by_owner
+        ("바로 발행" 버튼)를 명시적으로 눌러야 게재된다."""
         res = Res_WebPacketProtocol()
         err_type, _place = await self._load_place(user_info, place_id)
         if err_type != ErrorType.SUCCESS:
@@ -208,15 +208,13 @@ class PostService:
         await DB_SESSION_MNG.execute_lambda_run(
             [place_posts.DBType()], [lambda s: self.crud.update_body(s, pid, body)],
         )
-        await DB_SESSION_MNG.execute_lambda_run(
+        res.msg = "저장했습니다. 이메일 승인 링크를 눌러야 사이트에 반영됩니다."
-            [place_posts.DBType()], [lambda s: self.crud.approve(s, pid)],
-        )
-        await self._enqueue_build(pid, uuid.UUID(place_id))
-        res.msg = APPROVED
         return res
 
     async def approve_by_owner(self, user_info: UserInfo, place_id: str, post_id: str) -> Res_WebPacketProtocol:
-        """로그인 세션으로 바로 발행 — 고치지 않고 그대로. edit_by_owner 와 승인·빌드 경로는 같다."""
+        """로그인 세션으로 바로 발행 — 고치지 않고 그대로, 또는 방금 edit_by_owner 로 고친
+        그대로 승인한다(2026-09-21, 사장님 지시: "이메일 승인으로도 발행 가능하고
+        바로발행버튼으로도 발행 가능하도록"). 이메일 승인 링크와 별개의 두 번째 경로다."""
         res = Res_WebPacketProtocol()
         err_type, _place = await self._load_place(user_info, place_id)
         if err_type != ErrorType.SUCCESS:
@@ -224,13 +222,37 @@ class PostService:
             return res
 
         pid = uuid.UUID(post_id)
-        await DB_SESSION_MNG.execute_lambda_run(
+        await self._approve_and_publish(pid, uuid.UUID(place_id))
-            [place_posts.DBType()], [lambda s: self.crud.approve(s, pid)],
-        )
-        await self._enqueue_build(pid, uuid.UUID(place_id))
         res.msg = APPROVED
         return res
 
+    async def delete_by_owner(self, user_info: UserInfo, place_id: str, post_id: str) -> Res_WebPacketProtocol:
+        """소프트 삭제 — 상태 제한 없이 지운다. 이미 게재된 글이면 그 자리에서 빠지도록
+        재발행 잡까지 큐에 넣는다(그 외 상태는 사이트에 나간 적이 없어 재발행이 필요 없다)."""
+        res = Res_WebPacketProtocol()
+        err_type, _place = await self._load_place(user_info, place_id)
+        if err_type != ErrorType.SUCCESS:
+            res.result.SetResult(err_type)
+            return res
+
+        pid = uuid.UUID(post_id)
+        row = await DB_SESSION_MNG.execute_lambda(
+            place_posts.DBType(), DBWRType.DB_READ.value,
+            lambda s: self.crud.by_id(s, pid),
+        )
+        if row is None or str(row.place_id) != str(place_id):
+            res.result.SetResult(ErrorType.PLACE_NOT_FOUND)
+            return res
+
+        was_published = row.status == PostStatus.PUBLISHED.value
+        await DB_SESSION_MNG.execute_lambda_run(
+            [place_posts.DBType()], [lambda s: self.crud.delete(s, pid)],
+        )
+        if was_published:
+            await self._enqueue_build(pid, uuid.UUID(place_id))
+        res.msg = "삭제했습니다."
+        return res
+
     async def generate_range(self, user_info: UserInfo, place_id: str, start_date: date, end_date: date) -> Res_GenerateNow:
         """새벽 크론(04:10)을 기다리지 않고, 사장님이 고른 구간을 그 자리에서 채운다
         (2026-09-17, 사장님 지시: "지금 생성하기에서 시작이랑 끝 날짜를 정해야하지 않을까")."""
@@ -274,6 +296,34 @@ class PostService:
         res.msg = "만들었습니다."
         return res
 
+    _SEND_NOW_MSG = {
+        "MAIL_NOT_CONFIGURED": "메일 발송이 설정되어 있지 않습니다.",
+        "NO_VALID_EMAIL": "받을 이메일 주소가 올바르지 않습니다.",
+        "SEND_FAILED": "메일 발송에 실패했습니다.",
+    }
+
+    async def send_now(self, user_info: UserInfo, place_id: str) -> Res_WebPacketProtocol:
+        """빌더 화면의 '지금 발송하기' — 아침 9시 스윕을 기다리지 않고 이 업장의 오늘 몫을
+        바로 보낸다(2026-09-21, 사장님 요청). 보낼 게 없는 건 오류가 아니다 — generate_range
+        의 '이미 다 있거나 소재가 없다'와 같은 결의 안내로 끝낸다."""
+        res = Res_WebPacketProtocol()
+        err_type, _place = await self._load_place(user_info, place_id)
+        if err_type != ErrorType.SUCCESS:
+            res.result.SetResult(err_type)
+            return res
+
+        outcome = await blog_jobs.send_now_for_place(place_id)
+        if outcome["sent"]:
+            res.msg = "메일을 보냈습니다."
+            return res
+        reason = outcome.get("reason")
+        if reason == "NOTHING_DUE":
+            res.msg = "오늘 보낼 글이 없습니다."
+            return res
+        res.result.SetResult(ErrorType.INVALID_REQUEST_DATA)
+        res.msg = self._SEND_NOW_MSG.get(reason, "지금은 발송할 수 없습니다.")
+        return res
+
     async def _owner_user_id(self, place_id):
         def query(session):
             return session.execute(select(places.owner_user_id).where(places.place_id == place_id))
@@ -282,6 +332,28 @@ class PostService:
         row = result.first() if result is not None else None
         return row[0] if row else None
 
+    async def _approve_and_publish(self, post_id, place_id) -> None:
+        await DB_SESSION_MNG.execute_lambda_run(
+            [place_posts.DBType()], [lambda s: self.crud.approve(s, post_id)],
+        )
+        await self._enqueue_build(post_id, place_id)
+        await self._try_social_share(post_id, place_id)
+
+    async def _try_social_share(self, post_id, place_id) -> None:
+        """쓰레드 연동 — 실패해도 미니블로그 승인은 이미 끝난 뒤라 예외를 밖으로 던지지
+        않는다(2026-09-21, DECISIONS 7-1-2 개정)."""
+        try:
+            post = await DB_SESSION_MNG.execute_lambda(
+                place_posts.DBType(), DBWRType.DB_READ.value,
+                lambda s: self.crud.by_id(s, post_id),
+            )
+            owner_user_id = await self._owner_user_id(place_id)
+            if post is None or owner_user_id is None or not post.body:
+                return
+            await social_service.publish_reused_text(owner_user_id, place_id, post.body)
+        except Exception:
+            LOG.w(f"[blog] post={post_id} 쓰레드 연동 실패 — 미니블로그 승인은 유지")
+
     async def _enqueue_build(self, post_id, place_id) -> None:
         """게재 = 그 사이트 하나를 다시 굽는 것. 전체 재굽기가 아니다(docs/PUBLISH_VERSION.md).
 

solution/backend/services/social_service.py

@@ -19,7 +19,7 @@ from common.database.model.models import (
     place_facts,
     owner_social_accounts as Account,
 )
-from common.enums import SiteStatus, ErrorType, PUBLISHABLE_FACT_STATUSES
+from common.enums import SiteStatus, ErrorType, PUBLISHABLE_FACT_STATUSES, SocialProvider
 from crud.place_crud import PlaceCRUD
 from crud import social_crud as db
 from services import site_payload, social_account_service as accounts
@@ -580,3 +580,49 @@ async def run_post(job):
         except Exception:
             await set_failure(post_id, "POST_RESULT_UNKNOWN", status="UNKNOWN")
             raise SocialOutcomeUnknown("POST_RESULT_UNKNOWN") from None
+
+
+async def publish_reused_text(user_id, place_id, body: str):
+    """미니블로그 승인 문구를 그대로 쓰레드에 낸다 — 승인 자체가 발화 동의라 별도 승인을
+    또 묻지 않는다(2026-09-21, DECISIONS 7-1-2 개정: 미니블로그 문구를 그대로 재사용하는
+    경우에 한정한 예외). 연동 안 돼 있거나 조건 미달이면 조용히 None을 돌려준다 — 호출부가
+    실패로 취급하지 않는다."""
+    if not posting_enabled():
+        return None
+
+    async def run(s):
+        account = await accounts.account(s, user_id, SocialProvider.THREADS.value)
+        if not account or account.status != "linked":
+            return None
+        _place, site, url = await target(s, user_id, place_id)
+        full_text = f"{body}\n\n{url}"
+        row_id = (
+            await s.execute(
+                insert(Post)
+                .values(
+                    place_id=place_id,
+                    user_id=user_id,
+                    site_version_id=site.current_version_id,
+                    provider=SocialProvider.THREADS.value,
+                    account_id=account.account_id,
+                    link_url=url,
+                    body=full_text,
+                    status="APPROVED",
+                    decided_at=datetime.now(timezone.utc),
+                    decided_via="mini_blog",
+                )
+                .on_conflict_do_nothing(
+                    index_elements=["place_id", "site_version_id"],
+                    index_where=text("deleted=false"),
+                )
+                .returning(Post.post_id)
+            )
+        ).scalar_one_or_none()
+        if row_id:
+            await db.enqueue(s, row_id, 9)
+        return row_id
+
+    try:
+        return await db.transaction(run)
+    except HTTPException:
+        return None

solution/backend/tests/test_blog_owner.py

@@ -2,11 +2,14 @@
 
 ★ 이 파일이 지키는 것:
    - 로그인한 사장님은 자기 사업장의 글만 본다(남의 가게 글이 섞이면 안 된다)
-   - 아직 메일이 안 나간 REVIEWED 글도 로그인 화면에서 바로 고치고 승인할 수 있다
+   - 아직 메일이 안 나간 REVIEWED 글도 로그인 화면에서 바로 고칠 수 있다 — 단, 저장만
+     한다. 승인은 이메일 승인 링크 또는 로그인 "바로 발행" 버튼, 두 경로 중 하나를
+     명시적으로 눌러야 한다(2026-09-21, 사장님 지시: "이메일 승인으로도 발행 가능하고
+     바로발행버튼으로도 발행 가능하도록")
    - 여기서도 금칙 게이트는 그대로 탄다 — 로그인했다고 우회되지 않는다
 """
 import uuid
-from datetime import date, timedelta
+from datetime import date, timedelta, timezone
 
 from sqlalchemy import text
 
@@ -68,8 +71,9 @@ async def test_owner_cannot_see_someone_elses_posts(client, db_engine, auth_head
     assert body["posts"] == []
 
 
-async def test_owner_can_edit_and_approve_before_mail_goes_out(client, db_engine, auth_headers):
+async def test_owner_edit_saves_body_without_approving(client, db_engine, auth_headers):
-    """REVIEWED 글(아직 메일 안 나감)도 로그인 화면에서 바로 고쳐 승인할 수 있다."""
+    """로그인 화면에서 바로 고칠 수는 있지만, 저장만 한다 — 승인은 이메일 링크로만 일어난다
+    (2026-09-21, 사장님 지시: "승인되야 올라가도록 해야 한다")."""
     h = await auth_headers("blogowner4")
     place_id = await _place(client, h)
     post_id = await _seed_post(db_engine, place_id, status=PostStatus.REVIEWED)
@@ -78,7 +82,12 @@ async def test_owner_can_edit_and_approve_before_mail_goes_out(client, db_engine
     res = await client.put(f"/v1/place/{place_id}/post/{post_id}", headers=h, json={"body": new_body})
 
     assert res.json()["result"]["success"] is True
-    assert await _status(db_engine, post_id) == PostStatus.APPROVED.value
+    assert await _status(db_engine, post_id) == PostStatus.REVIEWED.value
+    async with db_engine.begin() as conn:
+        saved_body = (await conn.execute(
+            text("SELECT body FROM place_posts WHERE post_id = :id"), {"id": post_id},
+        )).scalar()
+    assert saved_body == new_body
 
 
 async def test_owner_edit_rejects_unverifiable_claims(client, db_engine, auth_headers):
@@ -95,12 +104,104 @@ async def test_owner_edit_rejects_unverifiable_claims(client, db_engine, auth_he
     assert await _status(db_engine, post_id) == PostStatus.REVIEWED.value
 
 
+async def test_owner_can_delete_a_post(client, db_engine, auth_headers):
+    h = await auth_headers("blogdel1")
+    place_id = await _place(client, h)
+    post_id = await _seed_post(db_engine, place_id, status=PostStatus.REVIEWED)
+
+    res = await client.delete(f"/v1/place/{place_id}/post/{post_id}", headers=h)
+
+    assert res.json()["result"]["success"] is True
+    async with db_engine.begin() as conn:
+        deleted = (await conn.execute(
+            text("SELECT deleted FROM place_posts WHERE post_id = :id"), {"id": post_id},
+        )).scalar()
+    assert deleted is True
+
+
+async def test_owner_cannot_delete_someone_elses_post(client, db_engine, auth_headers):
+    owner = await auth_headers("blogdel2")
+    other = await auth_headers("blogdel3")
+    place_id = await _place(client, owner)
+    post_id = await _seed_post(db_engine, place_id, status=PostStatus.REVIEWED)
+
+    res = await client.delete(f"/v1/place/{place_id}/post/{post_id}", headers=other)
+
+    assert res.json()["result"]["success"] is False
+    async with db_engine.begin() as conn:
+        deleted = (await conn.execute(
+            text("SELECT deleted FROM place_posts WHERE post_id = :id"), {"id": post_id},
+        )).scalar()
+    assert deleted is False
+
+
+async def test_deleting_a_post_frees_its_date_for_regeneration(client, db_engine, auth_headers, monkeypatch):
+    """삭제는 소프트 삭제라 (place_id, scheduled_date) 유니크가 풀린다 — 지운 날짜에
+    바로 다시 생성할 수 있어야 한다."""
+    from services import blog_service
+
+    async def fake_generate_one(*, place_name, region, topic_kind, material, used_topics, place_category):
+        return ("새로 만든 문구입니다. " + BODY, "gemini-test-model")
+
+    monkeypatch.setattr(blog_service, "generate_one", fake_generate_one)
+
+    h = await auth_headers("blogdel4")
+    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},
+        )
+    today = date.today()
+    post_id = await _seed_post(db_engine, place_id, status=PostStatus.REVIEWED, scheduled=today)
+
+    del_res = await client.delete(f"/v1/place/{place_id}/post/{post_id}", headers=h)
+    assert del_res.json()["result"]["success"] is True
+
+    gen_res = await client.post(
+        f"/v1/place/{place_id}/post/generate-one", headers=h, params={"date": today.isoformat()},
+    )
+
+    assert gen_res.json()["result"]["success"] is True
+
+
+async def test_deleting_a_published_post_enqueues_rebuild(client, db_engine, auth_headers):
+    h = await auth_headers("blogdel5")
+    place_id = await _place(client, h)
+    post_id = await _seed_post(db_engine, place_id, status=PostStatus.PUBLISHED)
+
+    res = await client.delete(f"/v1/place/{place_id}/post/{post_id}", headers=h)
+
+    assert res.json()["result"]["success"] is True
+    async with db_engine.begin() as conn:
+        payload = (await conn.execute(
+            text("SELECT payload FROM jobs WHERE job_type = :jt ORDER BY created_at DESC LIMIT 1"),
+            {"jt": JobType.BUILD.value},
+        )).scalar()
+    assert payload["place_id"] == place_id
+
+
+async def test_deleting_a_draft_post_does_not_enqueue_rebuild(client, db_engine, auth_headers):
+    h = await auth_headers("blogdel6")
+    place_id = await _place(client, h)
+    post_id = await _seed_post(db_engine, place_id, status=PostStatus.REVIEWED)
+
+    res = await client.delete(f"/v1/place/{place_id}/post/{post_id}", headers=h)
+
+    assert res.json()["result"]["success"] is True
+    async with db_engine.begin() as conn:
+        count = (await conn.execute(
+            text("SELECT count(*) FROM jobs WHERE job_type = :jt"), {"jt": JobType.BUILD.value},
+        )).scalar()
+    assert count == 0
+
+
 async def test_generate_now_creates_posts_for_published_site(client, db_engine, auth_headers, monkeypatch):
     """새벽 크론(04:10)을 기다리지 않고, 사장님이 고른 구간을 그 자리에서 채운다(2026-09-17,
     사장님 지시: "지금 생성하기에서 시작이랑 끝 날짜를 정해야하지 않을까") — 발행된 사이트일 때만."""
     from services import blog_service
 
-    async def fake_generate_one(*, place_name, region, topic_kind, material, used_topics):
+    async def fake_generate_one(*, place_name, region, topic_kind, material, used_topics, place_category):
         return ("테스트로 만든 문구입니다. " + BODY, "gemini-test-model")
 
     monkeypatch.setattr(blog_service, "generate_one", fake_generate_one)
@@ -109,8 +210,8 @@ async def test_generate_now_creates_posts_for_published_site(client, db_engine,
     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, status) VALUES (:sid, :pid, :st)"),
+            text("INSERT INTO sites (site_id, place_id, domain, status) VALUES (:sid, :pid, :dom, :st)"),
-            {"sid": uuid.uuid4(), "pid": place_id, "st": SiteStatus.PUBLISHED.value},
+            {"sid": uuid.uuid4(), "pid": place_id, "dom": f"blog-test-{uuid.uuid4().hex[:8]}", "st": SiteStatus.PUBLISHED.value},
         )
     start = date.today()
     end = start + timedelta(days=29)
@@ -140,6 +241,38 @@ 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_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):
+        return (f"테스트로 만든 문구입니다. {BODY}", "gemini-test-model")
+
+    monkeypatch.setattr(blog_service, "generate_one", fake_generate_one)
+
+    h = await auth_headers("bloglink1")
+    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},
+        )
+    today = date.today()
+
+    res = await client.post(
+        f"/v1/place/{place_id}/post/generate", headers=h,
+        params={"start": today.isoformat(), "end": today.isoformat()},
+    )
+    assert res.json()["created"] == 1
+
+    async with db_engine.begin() as conn:
+        body = (await conn.execute(
+            text("SELECT body FROM place_posts WHERE place_id = :pid"), {"pid": place_id},
+        )).scalar()
+    assert body == f"테스트로 만든 문구입니다. {BODY}"
+
+
 async def test_send_reviewed_only_mails_posts_due_today(client, db_engine, auth_headers, monkeypatch):
     """미래 날짜로 배정된 글은 그날이 오기 전엔 메일이 안 나간다."""
     from services import blog_jobs, mail_service
@@ -152,8 +285,8 @@ async def test_send_reviewed_only_mails_posts_due_today(client, db_engine, auth_
     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, status) VALUES (:sid, :pid, :st)"),
+            text("INSERT INTO sites (site_id, place_id, domain, status) VALUES (:sid, :pid, :dom, :st)"),
-            {"sid": uuid.uuid4(), "pid": place_id, "st": SiteStatus.PUBLISHED.value},
+            {"sid": uuid.uuid4(), "pid": place_id, "dom": f"blog-test-{uuid.uuid4().hex[:8]}", "st": SiteStatus.PUBLISHED.value},
         )
         await conn.execute(text("UPDATE users SET email = :e WHERE id = :id"), {"e": "owner@example.com", "id": "bloggen3"})
 
@@ -168,8 +301,218 @@ async def test_send_reviewed_only_mails_posts_due_today(client, db_engine, auth_
     assert await _status(db_engine, future_post) == PostStatus.REVIEWED.value
 
 
+async def test_send_reviewed_prefers_place_notify_email_over_account_email(client, db_engine, auth_headers, monkeypatch):
+    """places.notify_email 이 있으면 계정 로그인 이메일(users.email) 대신 그 주소로 보낸다
+    (2026-09-21, 사장님 요청 — 사장님 한 명이 사이트를 여러 개 가질 수 있어 업장별로
+    다른 담당자에게 보낼 수 있어야 한다)."""
+    from services import blog_jobs, mail_service
+
+    monkeypatch.setattr(mail_service, "is_configured", lambda: True)
+    sent_calls = []
+    monkeypatch.setattr(mail_service, "send", lambda **kwargs: sent_calls.append(kwargs) or True)
+
+    h = await auth_headers("bloggen4")
+    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},
+        )
+        await conn.execute(text("UPDATE users SET email = :e WHERE id = :id"), {"e": "owner@example.com", "id": "bloggen4"})
+
+    patch_res = await client.patch(f"/v1/place/{place_id}", headers=h, json={"notify_email": "manager@example.com"})
+    assert patch_res.json()["result"]["success"] is True
+
+    await _seed_post(db_engine, place_id, scheduled=date.today())
+    sent = await blog_jobs.send_reviewed()
+
+    assert sent == 1
+    assert sent_calls[0]["to"] == "manager@example.com"
+
+
+async def test_update_place_rejects_malformed_notify_email(client, db_engine, auth_headers):
+    h = await auth_headers("bloggen5")
+    place_id = await _place(client, h, name="잘못된이메일펜션")
+
+    res = await client.patch(f"/v1/place/{place_id}", headers=h, json={"notify_email": "not-an-email"})
+
+    assert res.json()["result"]["success"] is False
+
+
+async def test_send_now_mails_todays_due_post_immediately(client, db_engine, auth_headers, monkeypatch):
+    """빌더 화면의 '지금 발송하기' — 아침 9시 스윕을 기다리지 않고 이 업장의 오늘 몫을
+    바로 보낸다(2026-09-21, 사장님 요청)."""
+    from services import blog_jobs, mail_service
+
+    monkeypatch.setattr(mail_service, "is_configured", lambda: True)
+    sent_calls = []
+    monkeypatch.setattr(mail_service, "send", lambda **kwargs: sent_calls.append(kwargs) or True)
+
+    h = await auth_headers("bloggen6")
+    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},
+        )
+        await conn.execute(text("UPDATE users SET email = :e WHERE id = :id"), {"e": "owner6@example.com", "id": "bloggen6"})
+    post_id = await _seed_post(db_engine, place_id, scheduled=date.today())
+
+    res = await client.post(f"/v1/place/{place_id}/post/send-now", headers=h)
+
+    assert res.json()["result"]["success"] is True
+    assert len(sent_calls) == 1
+    assert sent_calls[0]["to"] == "owner6@example.com"
+    assert await _status(db_engine, post_id) == PostStatus.SENT.value
+
+
+async def test_send_now_is_a_noop_when_nothing_is_due(client, db_engine, auth_headers, monkeypatch):
+    from services import mail_service
+
+    monkeypatch.setattr(mail_service, "is_configured", lambda: True)
+    sent_calls = []
+    monkeypatch.setattr(mail_service, "send", lambda **kwargs: sent_calls.append(kwargs) or True)
+
+    h = await auth_headers("bloggen7")
+    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},
+        )
+    await _seed_post(db_engine, place_id, scheduled=date.today() + timedelta(days=3))
+
+    res = await client.post(f"/v1/place/{place_id}/post/send-now", headers=h)
+
+    body = res.json()
+    assert body["result"]["success"] is True
+    assert "없습니다" in body["msg"]
+    assert sent_calls == []
+
+
+async def test_send_now_fails_clearly_when_mail_is_not_configured(client, db_engine, auth_headers, monkeypatch):
+    from services import mail_service
+
+    monkeypatch.setattr(mail_service, "is_configured", lambda: False)
+
+    h = await auth_headers("bloggen8")
+    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},
+        )
+    await _seed_post(db_engine, place_id, scheduled=date.today())
+
+    res = await client.post(f"/v1/place/{place_id}/post/send-now", headers=h)
+
+    assert res.json()["result"]["success"] is False
+
+
+async def test_send_now_cannot_be_triggered_for_someone_elses_place(client, db_engine, auth_headers):
+    owner = await auth_headers("bloggen9")
+    other = await auth_headers("bloggen10")
+    place_id = await _place(client, owner, name="남의발송펜션")
+
+    res = await client.post(f"/v1/place/{place_id}/post/send-now", headers=other)
+
+    assert res.json()["result"]["success"] is False
+
+
+async def _day_pass_headers(db_engine, login_id: str) -> dict:
+    """메일의 '수정하려면' 링크가 주는 것과 같은 종류의 day-pass 토큰(2026-09-21,
+    사장님 지시: "메일 링크에서 들어와 수정한 후에는 승인할 수 있어야 한다")."""
+    from common.models.gmodel import UserInfo
+    from router.v1.validator.dependencies import CreateDayPassToken
+
+    async with db_engine.begin() as conn:
+        row = (await conn.execute(
+            text("SELECT user_id, role, token_version FROM users WHERE id = :id"), {"id": login_id},
+        )).first()
+    user_info = UserInfo(user_id=str(row[0]), id=login_id, role=row[1], token_version=row[2])
+    token = CreateDayPassToken(user_info)
+    return {"Authorization": f"Bearer {token}"}
+
+
+async def test_approve_by_owner_shares_to_threads_when_linked(client, db_engine, auth_headers, monkeypatch):
+    """'바로 발행' 버튼으로 승인하면, 쓰레드가 연동돼 있을 때 같은 문구가 쓰레드로도 나간다
+    (2026-09-21, 사장님 지시: "쓰레드에 연동되어 있으면 같이 업로드 되는 기능")."""
+    from services import post_service
+
+    calls = []
+
+    async def fake_publish_reused_text(user_id, place_id, body):
+        calls.append((user_id, place_id, body))
+        return uuid.uuid4()
+
+    monkeypatch.setattr(post_service.social_service, "publish_reused_text", fake_publish_reused_text)
+
+    h = await auth_headers("blogowner7")
+    place_id = await _place(client, h)
+    post_id = await _seed_post(db_engine, place_id, status=PostStatus.REVIEWED)
+
+    res = await client.post(f"/v1/place/{place_id}/post/{post_id}/approve", headers=h)
+
+    assert res.json()["result"]["success"] is True
+    assert len(calls) == 1
+    assert calls[0][2] == BODY
+    assert str(calls[0][1]) == place_id
+
+
+async def test_email_token_approve_also_shares_to_threads(client, db_engine, auth_headers, monkeypatch):
+    """이메일 GET 승인 경로도 '바로 발행' 버튼과 동일하게 쓰레드 연동을 태운다."""
+    from services import blog_service, post_service
+
+    calls = []
+
+    async def fake_publish_reused_text(user_id, place_id, body):
+        calls.append((user_id, place_id, body))
+        return uuid.uuid4()
+
+    monkeypatch.setattr(post_service.social_service, "publish_reused_text", fake_publish_reused_text)
+
+    h = await auth_headers("blogowner8")
+    place_id = await _place(client, h)
+    post_id = await _seed_post(db_engine, place_id, status=PostStatus.SENT)
+    token, token_hash, expires = blog_service.issue_token()
+    aware_expires = expires.replace(tzinfo=timezone.utc) if expires.tzinfo is None else expires
+    async with db_engine.begin() as conn:
+        await conn.execute(
+            text("UPDATE place_posts SET approve_token_hash=:h, token_expires_at=:e WHERE post_id=:id"),
+            {"h": token_hash, "e": aware_expires, "id": post_id},
+        )
+
+    res = await client.get(f"/v1/site/post/approve?t={token}")
+
+    assert res.status_code == 200
+    assert await _status(db_engine, post_id) == PostStatus.APPROVED.value
+    assert len(calls) == 1
+    assert calls[0][2] == BODY
+
+
+async def test_threads_share_failure_does_not_block_approval(client, db_engine, auth_headers, monkeypatch):
+    """쓰레드 연동이 예외를 던져도 미니블로그 승인 자체는 그대로 끝나야 한다."""
+    from services import post_service
+
+    async def failing_publish_reused_text(user_id, place_id, body):
+        raise RuntimeError("threads down")
+
+    monkeypatch.setattr(post_service.social_service, "publish_reused_text", failing_publish_reused_text)
+
+    h = await auth_headers("blogowner9")
+    place_id = await _place(client, h)
+    post_id = await _seed_post(db_engine, place_id, status=PostStatus.REVIEWED)
+
+    res = await client.post(f"/v1/place/{place_id}/post/{post_id}/approve", headers=h)
+
+    assert res.json()["result"]["success"] is True
+    assert await _status(db_engine, post_id) == PostStatus.APPROVED.value
+
+
 async def test_owner_can_publish_as_is_without_editing(client, db_engine, auth_headers):
-    """바로 발행 — 본문을 안 고쳐도 승인되고 재발행 잡이 걸린다."""
+    """'바로 발행' — 로그인 세션만으로, 본문을 안 고쳐도 승인되고 재발행 잡이 걸린다
+    (2026-09-21, 사장님 지시: "이메일 승인으로도 발행 가능하고 바로발행버튼으로도
+    발행 가능하도록") — 이메일 승인 링크와 별개의 두 번째 경로다."""
     h = await auth_headers("blogowner6")
     place_id = await _place(client, h)
     post_id = await _seed_post(db_engine, place_id, status=PostStatus.REVIEWED)
@@ -186,8 +529,8 @@ async def test_owner_can_publish_as_is_without_editing(client, db_engine, auth_h
 
 
 async def test_owner_cannot_publish_someone_elses_post(client, db_engine, auth_headers):
-    owner = await auth_headers("blogowner7")
+    owner = await auth_headers("blogowner6e")
-    other = await auth_headers("blogowner8")
+    other = await auth_headers("blogowner6f")
     place_id = await _place(client, owner)
     post_id = await _seed_post(db_engine, place_id, status=PostStatus.REVIEWED)
 
@@ -197,6 +540,43 @@ async def test_owner_cannot_publish_someone_elses_post(client, db_engine, auth_h
     assert await _status(db_engine, post_id) == PostStatus.REVIEWED.value
 
 
+async def test_daypass_session_can_approve_after_editing(client, db_engine, auth_headers):
+    """메일 '수정하려면' 링크(day-pass)로 들어와 고친 뒤에는, 다시 메일을 뒤지지 않고
+    그 자리에서 승인할 수 있어야 한다(2026-09-21, 사장님 지시)."""
+    h = await auth_headers("blogowner6b")
+    place_id = await _place(client, h)
+    post_id = await _seed_post(db_engine, place_id, status=PostStatus.REVIEWED)
+    day_pass = await _day_pass_headers(db_engine, "blogowner6b")
+    new_body = BODY.replace("빗소리", "새소리")
+
+    edit_res = await client.put(f"/v1/place/{place_id}/post/{post_id}", headers=day_pass, json={"body": new_body})
+    assert edit_res.json()["result"]["success"] is True
+    assert await _status(db_engine, post_id) == PostStatus.REVIEWED.value
+
+    approve_res = await client.post(f"/v1/place/{place_id}/post/{post_id}/approve", headers=day_pass)
+
+    assert approve_res.json()["result"]["success"] is True
+    assert await _status(db_engine, post_id) == PostStatus.APPROVED.value
+    async with db_engine.begin() as conn:
+        saved_body = (await conn.execute(
+            text("SELECT body FROM place_posts WHERE post_id = :id"), {"id": post_id},
+        )).scalar()
+    assert saved_body == new_body
+
+
+async def test_daypass_approve_cannot_be_used_for_someone_elses_place(client, db_engine, auth_headers):
+    owner = await auth_headers("blogowner6c")
+    place_id = await _place(client, owner)
+    post_id = await _seed_post(db_engine, place_id, status=PostStatus.REVIEWED)
+    await auth_headers("blogowner6d")
+    other_day_pass = await _day_pass_headers(db_engine, "blogowner6d")
+
+    res = await client.post(f"/v1/place/{place_id}/post/{post_id}/approve", headers=other_day_pass)
+
+    assert res.json()["result"]["success"] is False
+    assert await _status(db_engine, post_id) == PostStatus.REVIEWED.value
+
+
 async def test_generate_now_is_noop_for_unpublished_site(client, db_engine, auth_headers):
     """발행 전 사업장은 생성 스윕 대상이 아니다(blog_jobs._published_places) — 0건이어야 한다."""
     h = await auth_headers("bloggen2")
@@ -213,6 +593,36 @@ async def test_generate_now_is_noop_for_unpublished_site(client, db_engine, auth
     assert body["created"] == 0
 
 
+async def test_generate_now_is_noop_for_site_without_domain(client, db_engine, auth_headers, monkeypatch):
+    """domain 미확정(임시 주소) 사이트도 생성 스윕 대상이 아니다 — 쓰레드 연동 요구사항과
+    맞춘다(2026-09-21). PUBLISHED 여도 domain 이 없으면 0건이어야 한다. generate_one 을
+    실제로 성공하도록 목킹해 둬야 "그냥 LLM 이 설정 안 돼서 0건"과 구분된다."""
+    from services import blog_service
+
+    async def fake_generate_one(*, place_name, region, topic_kind, material, used_topics, place_category):
+        return ("테스트로 만든 문구입니다. " + BODY, "gemini-test-model")
+
+    monkeypatch.setattr(blog_service, "generate_one", fake_generate_one)
+
+    h = await auth_headers("bloggen17")
+    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, status) VALUES (:sid, :pid, :st)"),
+            {"sid": uuid.uuid4(), "pid": place_id, "st": SiteStatus.PUBLISHED.value},
+        )
+    start = date.today()
+
+    res = await client.post(
+        f"/v1/place/{place_id}/post/generate", headers=h,
+        params={"start": start.isoformat(), "end": start.isoformat()},
+    )
+
+    body = res.json()
+    assert body["result"]["success"] is True
+    assert body["created"] == 0
+
+
 async def test_generate_now_rejects_end_before_start(client, db_engine, auth_headers):
     h = await auth_headers("bloggen16")
     place_id = await _place(client, h, name="구간역순펜션")
@@ -310,7 +720,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):
+    async def fake_generate_one(*, place_name, region, topic_kind, material, used_topics, place_category):
         return ("테스트로 만든 문구입니다. " + BODY, "gemini-test-model")
 
     monkeypatch.setattr(blog_service, "generate_one", fake_generate_one)
@@ -319,8 +729,8 @@ async def test_generation_history_counts_by_batch(client, db_engine, auth_header
     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, status) VALUES (:sid, :pid, :st)"),
+            text("INSERT INTO sites (site_id, place_id, domain, status) VALUES (:sid, :pid, :dom, :st)"),
-            {"sid": uuid.uuid4(), "pid": place_id, "st": SiteStatus.PUBLISHED.value},
+            {"sid": uuid.uuid4(), "pid": place_id, "dom": f"blog-test-{uuid.uuid4().hex[:8]}", "st": SiteStatus.PUBLISHED.value},
         )
     start = date.today()
     generate_res = await client.post(
@@ -351,8 +761,8 @@ async def test_mail_has_one_click_approve_and_autologin_edit_links(client, db_en
     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, status) VALUES (:sid, :pid, :st)"),
+            text("INSERT INTO sites (site_id, place_id, domain, status) VALUES (:sid, :pid, :dom, :st)"),
-            {"sid": uuid.uuid4(), "pid": place_id, "st": SiteStatus.PUBLISHED.value},
+            {"sid": uuid.uuid4(), "pid": place_id, "dom": f"blog-test-{uuid.uuid4().hex[:8]}", "st": SiteStatus.PUBLISHED.value},
         )
         await conn.execute(
             text("UPDATE users SET email = :e WHERE id = :id"), {"e": "owner@example.com", "id": "bloggen11"},
@@ -369,11 +779,47 @@ async def test_mail_has_one_click_approve_and_autologin_edit_links(client, db_en
     assert "auto=" in mail_text
 
 
+async def test_mail_links_use_the_builder_app_origin_not_the_published_site_origin(
+    client, db_engine, auth_headers, monkeypatch,
+):
+    """수정·승인 링크는 빌더 앱(SOCIAL_APP_ORIGIN)으로 가야 한다 — 발행된 사이트 오리진
+    (site_payload.publish_origin, 로컬에선 solution-site 정적 서버 포트 80)으로 가면
+    404가 난다(2026-09-21 실측: 메일의 '수정하려면' 링크가 거기로 가서 404)."""
+    from services import blog_jobs, mail_service, site_payload
+
+    monkeypatch.setattr(mail_service, "is_configured", lambda: True)
+    sent_calls = []
+    monkeypatch.setattr(mail_service, "send", lambda **kwargs: sent_calls.append(kwargs) or True)
+    monkeypatch.setattr(site_payload, "publish_origin", lambda: "http://published-site-origin")
+    monkeypatch.setattr(
+        blog_jobs.social_config, "get",
+        lambda name, default="": "http://builder-app-origin" if name == "SOCIAL_APP_ORIGIN" else default,
+    )
+
+    h = await auth_headers("bloggen12")
+    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},
+        )
+        await conn.execute(text("UPDATE users SET email = :e WHERE id = :id"), {"e": "owner12@example.com", "id": "bloggen12"})
+    await _seed_post(db_engine, place_id, scheduled=date.today())
+
+    sent = await blog_jobs.send_reviewed()
+
+    assert sent == 1
+    mail_text = sent_calls[0]["text"]
+    assert "http://builder-app-origin/blog?placeId=" in mail_text
+    assert "http://builder-app-origin/v1/site/post/approve?t=" in mail_text
+    assert "published-site-origin" not in mail_text
+
+
 async def test_generate_one_fills_a_specific_empty_date(client, db_engine, auth_headers, monkeypatch):
     """사장님 지시: "개별적으로 새로 만들수있게 해줘" — 달력에서 빈 날짜 하나만 콕 집어 채운다."""
     from services import blog_service
 
-    async def fake_generate_one(*, place_name, region, topic_kind, material, used_topics):
+    async def fake_generate_one(*, place_name, region, topic_kind, material, used_topics, place_category):
         return ("테스트로 만든 문구입니다. " + BODY, "gemini-test-model")
 
     monkeypatch.setattr(blog_service, "generate_one", fake_generate_one)
@@ -382,8 +828,8 @@ async def test_generate_one_fills_a_specific_empty_date(client, db_engine, auth_
     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, status) VALUES (:sid, :pid, :st)"),
+            text("INSERT INTO sites (site_id, place_id, domain, status) VALUES (:sid, :pid, :dom, :st)"),
-            {"sid": uuid.uuid4(), "pid": place_id, "st": SiteStatus.PUBLISHED.value},
+            {"sid": uuid.uuid4(), "pid": place_id, "dom": f"blog-test-{uuid.uuid4().hex[:8]}", "st": SiteStatus.PUBLISHED.value},
         )
     target = date.today() + timedelta(days=3)
 
@@ -401,7 +847,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):
+    async def fake_generate_one(*, place_name, region, topic_kind, material, used_topics, place_category):
         return ("테스트로 만든 문구입니다. " + BODY, "gemini-test-model")
 
     monkeypatch.setattr(blog_service, "generate_one", fake_generate_one)
@@ -410,8 +856,8 @@ async def test_generate_one_fails_when_date_already_taken(client, db_engine, aut
     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, status) VALUES (:sid, :pid, :st)"),
+            text("INSERT INTO sites (site_id, place_id, domain, status) VALUES (:sid, :pid, :dom, :st)"),
-            {"sid": uuid.uuid4(), "pid": place_id, "st": SiteStatus.PUBLISHED.value},
+            {"sid": uuid.uuid4(), "pid": place_id, "dom": f"blog-test-{uuid.uuid4().hex[:8]}", "st": SiteStatus.PUBLISHED.value},
         )
     target = date.today() + timedelta(days=3)
     await _seed_post(db_engine, place_id, scheduled=target)

solution/backend/tests/test_social.py

@@ -157,39 +157,43 @@ async def test_expired_and_stale_sweep(client, auth_headers, db_engine):
 
 
 async def test_no_facts_no_paid_call(monkeypatch):
-    async def forbidden(*args, **kwargs):
+    from services.llm import provider as llm_provider
+
+    def forbidden():
         raise AssertionError("paid call")
 
-    monkeypatch.setattr(gemini_text, "call", forbidden)
+    monkeypatch.setattr(llm_provider, "active", forbidden)
     with pytest.raises(gemini_text.GeminiInvalidOutput, match="NO_GROUNDED_FACTS"):
         await gemini_text.generate_social_post("숙소", [], "https://example.com/s/stay")
 
 
 async def test_long_draft_regenerates(monkeypatch):
-    monkeypatch.setattr(gemini_text, "is_configured", lambda: True)
+    """generate_social_post 는 LLM_PROVIDER 추상화(services/llm/provider.py)를 탄다
+    (2026-09-21, Gemini 하드코딩 제거) — 여기서는 provider.active() 가 돌려주는 공급자
+    자체를 가짜로 바꿔 길이 초과 → 재요청 → 통과 흐름만 검증한다."""
+    from services.llm import provider as llm_provider
+    from services.llm.types import LlmResult, Usage
+
     bodies = iter(["가" * 501, "체크인은 15:00입니다."])
 
-    async def call(*args, **kwargs):
+    class _FakeLlm:
-        return {
+        __name__ = "services.llm.fake"
-            "candidates": [
+        DEFAULT_MODEL = "fake-model"
-                {
-                    "content": {
-                        "parts": [
-                            {
-                                "text": json.dumps(
-                                    {
-                                        "body": next(bodies),
-                                        "fact_keys": ["check_in_time"],
-                                    }
-                                )
-                            }
-                        ]
-                    }
-                }
-            ]
-        }
 
-    monkeypatch.setattr(gemini_text, "call", call)
+        @staticmethod
+        def is_configured():
+            return True
+
+        @staticmethod
+        async def generate(client, model, *, prompt, response_schema=None, temperature=0.2, max_retries=0, images=None):
+            body = next(bodies)
+            return LlmResult(
+                json={"body": body, "fact_keys": ["check_in_time"]},
+                text=json.dumps({"body": body, "fact_keys": ["check_in_time"]}),
+                usage=Usage(input_tokens=0, output_tokens=0),
+            )
+
+    monkeypatch.setattr(llm_provider, "active", lambda: _FakeLlm())
     result = await gemini_text.generate_social_post(
         "숙소",
         [gemini_text.FactInput(key="check_in_time", label="체크인", value="15:00")],
@@ -495,3 +499,103 @@ async def test_oauth_roundtrip_saves_encrypted_account(db_engine, monkeypatch):
             "SELECT status, access_token FROM owner_social_accounts WHERE user_id = :u"),
             {"u": user_id})).first()
     assert after.status == "revoked" and after.access_token is None, "해제해도 토큰이 남아 있다"
+
+
+async def test_publish_reused_text_skips_when_no_threads_account(db_engine, auth_headers, client):
+    from services import social_service as svc
+
+    h, pid, uid, v = await seed(client, auth_headers, db_engine)
+    result = await svc.publish_reused_text(uid, pid, "오늘도 마당이 조용합니다.")
+    assert result is None
+    async with db_engine.begin() as c:
+        assert (
+            await c.execute(text("SELECT count(*) FROM place_social_posts WHERE place_id=:p"), {"p": pid})
+        ).scalar_one() == 0
+
+
+async def test_publish_reused_text_skips_when_posting_disabled(db_engine, auth_headers, client, monkeypatch):
+    from services import social_service as svc
+
+    monkeypatch.setattr(svc, "posting_enabled", lambda: False)
+    h, pid, uid, v = await seed(client, auth_headers, db_engine)
+    async with db_engine.begin() as c:
+        await c.execute(
+            text(
+                "INSERT INTO owner_social_accounts(account_id,user_id,provider,provider_user_id,handle,profile_url,status) "
+                "VALUES (:a,:u,2,'22','host','https://www.threads.com/@host','linked')"
+            ),
+            {"a": uuid.uuid4(), "u": uid},
+        )
+    result = await svc.publish_reused_text(uid, pid, "오늘도 마당이 조용합니다.")
+    assert result is None
+    async with db_engine.begin() as c:
+        assert (
+            await c.execute(text("SELECT count(*) FROM place_social_posts WHERE place_id=:p"), {"p": pid})
+        ).scalar_one() == 0
+
+
+async def test_publish_reused_text_skips_when_domain_not_fixed(db_engine, auth_headers, client, monkeypatch):
+    from services import social_service as svc
+
+    monkeypatch.setattr(svc, "posting_enabled", lambda: True)
+    h, pid, uid, v = await seed(client, auth_headers, db_engine)
+    async with db_engine.begin() as c:
+        await c.execute(text("UPDATE sites SET domain=NULL WHERE place_id=:p"), {"p": pid})
+        await c.execute(
+            text(
+                "INSERT INTO owner_social_accounts(account_id,user_id,provider,provider_user_id,handle,profile_url,status) "
+                "VALUES (:a,:u,2,'22','host','https://www.threads.com/@host','linked')"
+            ),
+            {"a": uuid.uuid4(), "u": uid},
+        )
+    result = await svc.publish_reused_text(uid, pid, "오늘도 마당이 조용합니다.")
+    assert result is None
+
+
+async def test_publish_reused_text_inserts_approved_post_with_link_and_enqueues_run_post(
+    db_engine, auth_headers, client, monkeypatch
+):
+    from services import social_service as svc
+
+    monkeypatch.setattr(svc, "posting_enabled", lambda: True)
+    h, pid, uid, v = await seed(client, auth_headers, db_engine)
+    account_id = uuid.uuid4()
+    async with db_engine.begin() as c:
+        await c.execute(
+            text(
+                "INSERT INTO owner_social_accounts(account_id,user_id,provider,provider_user_id,handle,profile_url,status) "
+                "VALUES (:a,:u,2,'22','host','https://www.threads.com/@host','linked')"
+            ),
+            {"a": account_id, "u": uid},
+        )
+
+    result = await svc.publish_reused_text(uid, pid, "오늘도 마당이 조용합니다.")
+
+    assert result is not None
+    expected_url = svc.site_payload.publish_origin() + "/s/social-stay"
+    async with db_engine.begin() as c:
+        row = (
+            await c.execute(
+                text(
+                    "SELECT status, body, link_url, decided_via, account_id FROM place_social_posts WHERE post_id=:p"
+                ),
+                {"p": result},
+            )
+        ).first()
+        assert row.status == "APPROVED"
+        assert row.body == f"오늘도 마당이 조용합니다.\n\n{expected_url}"
+        assert row.link_url == expected_url
+        assert row.decided_via == "mini_blog"
+        assert row.account_id == account_id
+        job_count = (
+            await c.execute(text("SELECT count(*) FROM jobs WHERE job_type=9"))
+        ).scalar_one()
+        assert job_count == 1
+
+    # 같은 site_version 에 두 번째 호출 — 유니크 인덱스가 중복 삽입을 막는다
+    second = await svc.publish_reused_text(uid, pid, "다른 문구")
+    assert second is None
+    async with db_engine.begin() as c:
+        assert (
+            await c.execute(text("SELECT count(*) FROM place_social_posts WHERE place_id=:p"), {"p": pid})
+        ).scalar_one() == 1