feat(agent-office): realestate on_new_matches + /notify endpoint

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

agent-office/app/agents/realestate.py

@@ -1,89 +1,68 @@
-import asyncio
 from .base import BaseAgent
 from ..db import create_task, update_task_status, add_log
 from .. import service_proxy
-from .. import telegram_bot
+from ..telegram import messaging
+from ..telegram.realestate_message import format_realestate_matches, build_match_keyboard
 
 
 class RealestateAgent(BaseAgent):
     """부동산 청약 에이전트.
 
-    매일 09:15 자동 실행: realestate-lab의 수집을 트리거하고
-    신규 매칭 결과를 텔레그램으로 푸시 (승인 없는 리포트형).
+    realestate-lab이 신규 매칭 발견 시 /realestate/notify로 push해 트리거됨.
+    on_new_matches가 메인 진입점. on_schedule은 사용하지 않음(cron 폐기).
     """
 
     agent_id = "realestate"
     display_name = "청약 애널리스트"
 
-    async def on_schedule(self) -> None:
-        if self.state not in ("idle", "break"):
-            return
+    async def on_new_matches(self, matches: list[dict]) -> dict:
+        """신규 매칭 N건을 텔레그램 1통으로 푸시.
+        성공 시 sent_ids 반환 → realestate-lab이 notified_at 마킹.
+        실패 시 sent=0, sent_ids=[] 반환 → 다음 사이클 재시도.
+        """
+        if not matches:
+            return {"sent": 0, "sent_ids": []}
 
-        task_id = create_task(self.agent_id, "daily_match_report", {})
-        await self.transition("working", "청약 공고 수집 중", task_id)
+        task_id = create_task(self.agent_id, "notify_matches", {"count": len(matches)})
 
         try:
-            collect = await service_proxy.realestate_collect()
-            new_count = collect.get("new_count", 0) or 0
+            text = format_realestate_matches(matches)
+            keyboard = build_match_keyboard(matches)
+            await self.transition("reporting", f"매칭 {len(matches)}건 알림", task_id)
 
-            await self.transition("working", "신규 매칭 조회 중", task_id)
-            matches = await service_proxy.realestate_matches(limit=20)
-            dashboard = await service_proxy.realestate_dashboard()
-
-            await self.transition("reporting", "리포트 전송 중", task_id)
-
-            if not matches:
-                body = (
-                    f"수집된 신규 공고: {new_count}건\n"
-                    f"진행 중 공고: {dashboard.get('active_count', 0)}건\n"
-                    f"신규 매칭: 없음"
-                )
-            else:
-                lines = [
-                    f"📌 수집 {new_count}건 / 매칭 {len(matches)}건",
-                    "",
-                ]
-                for m in matches[:5]:
-                    title = m.get("title") or m.get("announcement_title") or "(제목 없음)"
-                    region = m.get("region") or ""
-                    score = m.get("match_score") or m.get("score") or ""
-                    lines.append(f"• [{region}] {title} (매칭 {score})")
-                if len(matches) > 5:
-                    lines.append(f"… 외 {len(matches) - 5}건")
-                body = "\n".join(lines)
-
-            tg = await telegram_bot.send_task_result(
-                self.agent_id,
-                "🏢 [청약 에이전트] 오늘의 매칭 리포트",
-                body,
-            )
-
-            # 확인한 매칭 read 처리
-            for m in matches[:5]:
-                mid = m.get("id")
-                if mid:
-                    try:
-                        await service_proxy.realestate_mark_read(int(mid))
-                    except Exception:
-                        pass
+            tg = await messaging.send_raw(text, reply_markup=keyboard)
+            if not tg.get("ok"):
+                update_task_status(task_id, "failed", {"error": tg.get("description")})
+                await self.transition("idle", "알림 실패")
+                return {"sent": 0, "sent_ids": [], "error": tg.get("description")}
 
+            sent_ids = [m["id"] for m in matches if "id" in m]
             update_task_status(task_id, "succeeded", {
-                "new_count": new_count,
-                "match_count": len(matches),
-                "telegram_sent": tg.get("ok", False),
+                "sent": len(matches),
                 "telegram_message_id": tg.get("message_id"),
             })
-            await self.transition("idle", f"매칭 {len(matches)}건")
-
+            await self.transition("idle", f"매칭 {len(matches)}건 알림 완료")
+            return {
+                "sent": len(matches),
+                "sent_ids": sent_ids,
+                "message_id": tg.get("message_id"),
+            }
         except Exception as e:
-            add_log(self.agent_id, f"Realestate report failed: {e}", "error", task_id)
+            add_log(self.agent_id, f"on_new_matches failed: {e}", "error", task_id)
             update_task_status(task_id, "failed", {"error": str(e)})
             await self.transition("idle", f"오류: {e}")
+            return {"sent": 0, "sent_ids": [], "error": str(e)}
 
     async def on_command(self, command: str, params: dict) -> dict:
         if command == "fetch_matches":
-            await self.on_schedule()
-            return {"ok": True, "message": "매칭 리포트 시작"}
+            try:
+                matches = await service_proxy.realestate_matches(limit=20)
+                if not matches:
+                    return {"ok": True, "message": "매칭 없음"}
+                result = await self.on_new_matches(matches)
+                return {"ok": True, "result": result}
+            except Exception as e:
+                return {"ok": False, "message": str(e)}
 
         if command == "dashboard":
             try:

agent-office/app/main.py

@@ -180,3 +180,22 @@ def conversation_stats(days: int = 7):
 @app.get("/api/agent-office/activity")
 def activity_feed(limit: int = 50, offset: int = 0):
     return get_activity_feed(limit, offset)
+
+
+# --- Realestate Agent Push Endpoint ---
+
+from pydantic import BaseModel
+from typing import List, Dict, Any
+
+
+class RealestateNotifyBody(BaseModel):
+    matches: List[Dict[str, Any]]
+
+
+@app.post("/api/agent-office/realestate/notify")
+async def realestate_notify(body: RealestateNotifyBody):
+    agent = get_agent("realestate")
+    if agent is None:
+        from fastapi import HTTPException
+        raise HTTPException(status_code=503, detail="RealestateAgent not initialized")
+    return await agent.on_new_matches(body.matches)

agent-office/tests/test_realestate_agent.py

@@ -0,0 +1,97 @@
+import os
+import sys
+import tempfile
+
+_TMP = tempfile.mktemp(suffix=".db")
+os.environ["AGENT_OFFICE_DB_PATH"] = _TMP
+
+sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+
+
+import asyncio
+from unittest.mock import AsyncMock, patch
+import pytest
+
+
+@pytest.fixture(autouse=True)
+def _init_db():
+    import gc
+    gc.collect()
+    if os.path.exists(_TMP):
+        os.remove(_TMP)
+    from app.db import init_db
+    init_db()
+    yield
+    gc.collect()
+
+
+def test_on_new_matches_returns_empty_when_no_matches():
+    from app.agents.realestate import RealestateAgent
+
+    agent = RealestateAgent()
+    result = asyncio.run(agent.on_new_matches([]))
+    assert result == {"sent": 0, "sent_ids": []}
+
+
+def test_on_new_matches_sends_telegram_and_returns_ids():
+    from app.agents.realestate import RealestateAgent
+    from app.telegram import messaging
+
+    matches = [{
+        "id": 7, "match_score": 80, "house_nm": "단지A",
+        "region_name": "서울특별시", "district": "강남구",
+        "receipt_start": "2026-05-01", "receipt_end": "2026-05-05",
+        "match_reasons": [], "eligible_types": [], "pblanc_url": "https://x.test/7",
+    }]
+
+    fake_send = AsyncMock(return_value={"ok": True, "message_id": 123})
+    with patch.object(messaging, "send_raw", fake_send):
+        agent = RealestateAgent()
+        result = asyncio.run(agent.on_new_matches(matches))
+
+    assert result["sent"] == 1
+    assert result["sent_ids"] == [7]
+    assert result["message_id"] == 123
+    fake_send.assert_awaited_once()
+    args, kwargs = fake_send.call_args
+    text = args[0]
+    assert "단지A" in text
+
+
+def test_on_new_matches_telegram_failure_returns_zero():
+    from app.agents.realestate import RealestateAgent
+    from app.telegram import messaging
+
+    matches = [{
+        "id": 8, "match_score": 80, "house_nm": "단지B",
+        "region_name": "서울", "district": "송파구",
+        "receipt_start": "", "receipt_end": "",
+        "match_reasons": [], "eligible_types": [], "pblanc_url": "",
+    }]
+
+    fake_send = AsyncMock(return_value={"ok": False, "description": "401"})
+    with patch.object(messaging, "send_raw", fake_send):
+        agent = RealestateAgent()
+        result = asyncio.run(agent.on_new_matches(matches))
+
+    assert result["sent"] == 0
+    assert result["sent_ids"] == []
+    assert "error" in result
+
+
+def test_endpoint_calls_agent_on_new_matches():
+    from fastapi.testclient import TestClient
+    from app.main import app
+    from app.agents.realestate import RealestateAgent
+
+    fake = AsyncMock(return_value={"sent": 1, "sent_ids": [99], "message_id": 1})
+    with patch.object(RealestateAgent, "on_new_matches", fake):
+        with TestClient(app) as client:
+            resp = client.post(
+                "/api/agent-office/realestate/notify",
+                json={"matches": [{"id": 99, "match_score": 80}]},
+            )
+    assert resp.status_code == 200
+    body = resp.json()
+    assert body["sent"] == 1
+    assert body["sent_ids"] == [99]