diff --git a/realestate-lab/app/notifier.py b/realestate-lab/app/notifier.py
new file mode 100644
index 0000000..94ab940
--- /dev/null
+++ b/realestate-lab/app/notifier.py
@@ -0,0 +1,45 @@
+"""신규 매칭을 agent-office로 push하여 텔레그램 알림을 트리거한다."""
+import os
+import logging
+import requests
+
+from .db import get_profile, get_unnotified_matches, mark_matches_notified
+
+logger = logging.getLogger("realestate-lab")
+
+AGENT_OFFICE_URL = os.getenv("AGENT_OFFICE_URL", "http://agent-office:8000")
+NOTIFY_TIMEOUT_SECONDS = int(os.getenv("REALESTATE_NOTIFY_TIMEOUT", "15"))
+
+
+def notify_new_matches() -> dict:
+    """프로필의 임계값을 통과한 미알림 매칭을 agent-office로 push한다.
+
+    응답이 200이고 sent_ids가 비어있지 않으면 해당 IDs의 notified_at을 마킹.
+    실패 시 마킹하지 않아 다음 사이클에서 재시도된다.
+    """
+    profile = get_profile()
+    if not profile:
+        return {"sent": 0, "skipped": "no_profile"}
+
+    if not profile.get("notify_enabled"):
+        return {"sent": 0, "skipped": "notify_disabled"}
+
+    threshold = profile.get("min_match_score") or 70
+    matches = get_unnotified_matches(threshold)
+    if not matches:
+        return {"sent": 0}
+
+    url = f"{AGENT_OFFICE_URL}/api/agent-office/realestate/notify"
+    try:
+        resp = requests.post(url, json={"matches": matches}, timeout=NOTIFY_TIMEOUT_SECONDS)
+        resp.raise_for_status()
+        body = resp.json()
+    except requests.RequestException as e:
+        logger.error("agent-office push 실패: %s", e)
+        return {"sent": 0, "error": str(e)}
+
+    sent_ids = body.get("sent_ids") or []
+    if sent_ids:
+        mark_matches_notified(sent_ids)
+        logger.info("알림 송신: %d건", len(sent_ids))
+    return body
diff --git a/realestate-lab/tests/test_notifier.py b/realestate-lab/tests/test_notifier.py
new file mode 100644
index 0000000..2c29127
--- /dev/null
+++ b/realestate-lab/tests/test_notifier.py
@@ -0,0 +1,85 @@
+from unittest.mock import patch, MagicMock
+
+
+def _seed_profile_and_match(score, notify_enabled=True, threshold=70):
+    from app.db import _conn, upsert_profile
+    upsert_profile({
+        "name": "u",
+        "notify_enabled": notify_enabled,
+        "min_match_score": threshold,
+    })
+    with _conn() as conn:
+        conn.execute("""
+            INSERT INTO announcements (house_manage_no, pblanc_no, house_nm, status, source)
+            VALUES ('NF1', '01', '단지', '청약중', 'manual')
+        """)
+        ann_id = conn.execute("SELECT id FROM announcements WHERE house_manage_no='NF1'").fetchone()["id"]
+        conn.execute("""
+            INSERT INTO match_results (announcement_id, model_id, match_score, match_reasons, eligible_types, is_new)
+            VALUES (?, NULL, ?, '[]', '[]', 1)
+        """, (ann_id, score))
+        match_id = conn.execute("SELECT id FROM match_results WHERE announcement_id=?", (ann_id,)).fetchone()["id"]
+    return match_id
+
+
+def test_notify_skips_when_disabled():
+    from app import notifier
+    _seed_profile_and_match(score=80, notify_enabled=False)
+    with patch.object(notifier, "requests") as r:
+        result = notifier.notify_new_matches()
+    assert r.post.call_count == 0
+    assert result["sent"] == 0
+    assert result.get("skipped") == "notify_disabled"
+
+
+def test_notify_filters_below_threshold():
+    from app import notifier
+    _seed_profile_and_match(score=60, threshold=70)
+    with patch.object(notifier, "requests") as r:
+        result = notifier.notify_new_matches()
+    assert r.post.call_count == 0
+    assert result["sent"] == 0
+
+
+def test_notify_pushes_and_marks_notified():
+    from app import notifier
+    from app.db import _conn
+
+    match_id = _seed_profile_and_match(score=80, threshold=70)
+
+    fake_resp = MagicMock()
+    fake_resp.json.return_value = {"sent": 1, "sent_ids": [match_id]}
+    fake_resp.raise_for_status.return_value = None
+
+    with patch.object(notifier.requests, "post", return_value=fake_resp) as post:
+        result = notifier.notify_new_matches()
+
+    assert post.call_count == 1
+    args, kwargs = post.call_args
+    assert "/api/agent-office/realestate/notify" in args[0]
+    assert kwargs["json"]["matches"][0]["id"] == match_id
+
+    with _conn() as conn:
+        row = conn.execute("SELECT notified_at FROM match_results WHERE id=?", (match_id,)).fetchone()
+    assert row["notified_at"] is not None
+    assert result["sent"] == 1
+
+
+def test_notify_does_not_mark_on_failure():
+    from app import notifier
+    from app.db import _conn
+    import requests as real_requests
+
+    match_id = _seed_profile_and_match(score=80, threshold=70)
+
+    def boom(*a, **k):
+        raise real_requests.RequestException("agent-office down")
+
+    with patch.object(notifier.requests, "post", side_effect=boom):
+        result = notifier.notify_new_matches()
+
+    with _conn() as conn:
+        row = conn.execute("SELECT notified_at FROM match_results WHERE id=?", (match_id,)).fetchone()
+    assert row["notified_at"] is None
+    assert result["sent"] == 0
+    assert "error" in result