결론

기존 Google Home TTS 루틴은 유지했다. tts_morning.sh, tts_weekly.sh, 기존 com.openclaw.morning-routine.plist, com.openclaw.weekly-report.plist는 수정하지 않았다.
Telegram report 토픽용 신규 wrapper 2개와 공통 Python wrapper 1개를 추가했다.
신규 LaunchAgent plist 2개를 생성했다.
수동 실행은 wrapper 생성/본문 구성/sector report 라우팅까지 도달했으나, 현재 Codex 실행 환경에서 api.telegram.org DNS/네트워크가 막혀 실제 Telegram API 발송은 실패했다. 따라서 message_id는 아직 없다.
launchctl bootstrap도 현재 세션에서 Bootstrap failed: 5: Input/output error로 실패했다. 파일은 정상 생성·lint 완료, 일반 터미널/정상 launchd 환경에서 bootstrap 필요.

현재 TTS 루틴 분석

morning-routine

대상: /Users/ron/.hermes/skills/smart-home/scripts/tts_morning.sh

역할: - Home Assistant에서 현재 날씨/예보 조회 - Hermes 운영 DB에서 high/urgent 대기 작업 3개 조회 - yfinance + RSS + LLM으로 미국 시황 구어체 요약 시도 - 평일 조명/커튼 제어 - Google Home (192.168.0.8, friendly_name=거실)에 TTS 캐스팅 - 이후 KBS 1라디오 또는 주말 클래식FM 스트림 재생

현재 plist: - /Users/ron/Library/LaunchAgents/com.openclaw.morning-briefing-telegram.plist가 아니라 기존 /Users/ron/Library/LaunchAgents/com.openclaw.morning-routine.plist - 06:30 실행 - ProgramArguments: /bin/bash /Users/ron/.hermes/skills/smart-home/scripts/tts_morning.sh

주의: - 이 세션 직접 실행 로그에는 zeroconf 소켓 권한 제한으로 TTS 캐스팅 실패가 있었지만, 스크립트 자체는 [OK] 아침 루틴 완료까지 진행한다. 이건 Codex sandbox 네트워크/소켓 제한의 영향으로 보인다.

weekly-report

대상: /Users/ron/.hermes/skills/smart-home/scripts/tts_weekly.sh

역할: - Hermes 운영 DB에서 최근 7일 완료 작업 수 / 현재 queued 수 조회 - agent_kpi_daily 평균 KPI 조회 - yfinance로 S&P500/NASDAQ 5일 변화율 조회 - 최근 7일 reports/ideas/scored_discoveries_*.jsonl 발견 수 집계 - Google Home TTS로 주간 리포트 송출

현재 plist: - 기존 /Users/ron/Library/LaunchAgents/com.openclaw.weekly-report.plist - 일요일 20:00 실행 - ProgramArguments: /bin/bash /Users/ron/.hermes/skills/smart-home/scripts/tts_weekly.sh

morning_briefing.py 분석

대상: /Users/ron/.hermes/workspace/scripts/pipeline/morning_briefing.py

확인한 구조: - _load_fed_liquidity() → ~/.hermes/workspace/memory/fed-liquidity/latest.json - _fed_liquidity_oneliner() → 시장 지표 섹션에 한 줄 삽입 - _summarize_fed_liquidity() → 유동성 상세 섹션 생성 - collect_all_data() 데이터 소스: - market-indicators - strategy-flow - commodity-alerts - geopolitical - china-macro - cu-research - hypotheses / experiment-results / filtered-ideas - analyst-pm / analyst-macro - sector-news / oil-supply / sector-cycles - fed-liquidity - economic-calendar - vault methodology/judgments/structured context - send_telegram()은 현재 send_dm_chunked()를 사용한다. 이 경로는 현재 정책상 DM이 아니라 알림센터 일반 경로로 redirect될 수 있어, 해리가 원하는 report topic 보장에는 맞지 않는다.

따라서 morning_briefing.py 원본은 건드리지 않고, 별도 Telegram report-topic wrapper를 만들었다.

신규 설계

생성 파일

종류	경로	역할
공통 Python wrapper	`/Users/ron/.hermes/workspace/scripts/pipeline/telegram_briefing_wrapper.py`	morning/weekly 메시지 생성 + report topic 발송
morning shell wrapper	`/Users/ron/.hermes/workspace/scripts/pipeline/morning_briefing_wrapper.sh`	launchd에서 Python wrapper 호출
weekly shell wrapper	`/Users/ron/.hermes/workspace/scripts/pipeline/weekly_briefing_wrapper.sh`	launchd에서 Python wrapper 호출
morning plist	`/Users/ron/Library/LaunchAgents/com.openclaw.morning-briefing-telegram.plist`	매일 06:45 report topic 전송
weekly plist	`/Users/ron/Library/LaunchAgents/com.openclaw.weekly-briefing-telegram.plist`	월요일 07:00 report topic 전송

발송 방식

~/.hermes/sector-group.json 확인 결과:
chat_id: -1003522748967
report topic id: 8
wrapper는 동일 설정을 읽어 report topic으로 보낸다.
shared.telegram.send_sector()는 bool만 반환해서 message_id를 기록할 수 없다.
그래서 wrapper는 send_sector(report)와 같은 sector config를 쓰되, shared.telegram._send_raw(... topic_id=8)로 발송하여 성공 시 message_id를 받을 수 있게 했다.
또한 로그에는 [SECTOR_TRACE] send_sector(report) 형태를 남긴다.
현재 Codex 환경의 DNS 실패에 대비해 OpenClaw CLI fallback도 추가했다.

신규 plist

morning Telegram plist

<?xml version="1.0" encoding="UTF-8"?>
<!DOCTYPE plist PUBLIC "-//Apple//DTD PLIST 1.0//EN" "http://www.apple.com/DTDs/PropertyList-1.0.dtd">
<plist version="1.0">
<dict>
    <key>Label</key>
    <string>com.openclaw.morning-briefing-telegram</string>
    <key>ProgramArguments</key>
    <array>
        <string>/bin/bash</string>
        <string>/Users/ron/.hermes/workspace/scripts/pipeline/morning_briefing_wrapper.sh</string>
    </array>
    <key>WorkingDirectory</key>
    <string>/Users/ron/.hermes/workspace</string>
    <key>StartCalendarInterval</key>
    <dict>
        <key>Hour</key>
        <integer>6</integer>
        <key>Minute</key>
        <integer>45</integer>
    </dict>
    <key>StandardOutPath</key>
    <string>/Users/ron/.hermes/logs/morning-briefing-telegram.log</string>
    <key>StandardErrorPath</key>
    <string>/Users/ron/.hermes/logs/morning-briefing-telegram.err.log</string>
    <key>EnvironmentVariables</key>
    <dict>
        <key>HOME</key>
        <string>/Users/ron</string>
        <key>PATH</key>
        <string>/opt/homebrew/bin:/usr/local/bin:/usr/bin:/bin:/usr/sbin:/sbin</string>
    </dict>
</dict>
</plist>

weekly Telegram plist

<?xml version="1.0" encoding="UTF-8"?>
<!DOCTYPE plist PUBLIC "-//Apple//DTD PLIST 1.0//EN" "http://www.apple.com/DTDs/PropertyList-1.0.dtd">
<plist version="1.0">
<dict>
    <key>Label</key>
    <string>com.openclaw.weekly-briefing-telegram</string>
    <key>ProgramArguments</key>
    <array>
        <string>/bin/bash</string>
        <string>/Users/ron/.hermes/workspace/scripts/pipeline/weekly_briefing_wrapper.sh</string>
    </array>
    <key>WorkingDirectory</key>
    <string>/Users/ron/.hermes/workspace</string>
    <key>StartCalendarInterval</key>
    <dict>
        <key>Weekday</key>
        <integer>1</integer>
        <key>Hour</key>
        <integer>7</integer>
        <key>Minute</key>
        <integer>0</integer>
    </dict>
    <key>StandardOutPath</key>
    <string>/Users/ron/.hermes/logs/weekly-briefing-telegram.log</string>
    <key>StandardErrorPath</key>
    <string>/Users/ron/.hermes/logs/weekly-briefing-telegram.err.log</string>
    <key>EnvironmentVariables</key>
    <dict>
        <key>HOME</key>
        <string>/Users/ron</string>
        <key>PATH</key>
        <string>/opt/homebrew/bin:/usr/local/bin:/usr/bin:/bin:/usr/sbin:/sbin</string>
    </dict>
</dict>
</plist>

검증:

plutil -lint /Users/ron/Library/LaunchAgents/com.openclaw.morning-briefing-telegram.plist /Users/ron/Library/LaunchAgents/com.openclaw.weekly-briefing-telegram.plist
=> OK

수동 검증 결과

dry-run

morning no-LLM dry-run


## 수급
수급 데이터 없음

## 매크로/PM
매크로 레짐: Recovery (확신도 57%)
요약: 에너지 복합체 전면 급락이 핵심 변화. GPR 304 극단에도 시장은 수요 약화 반영. 산업금속(Cu z=2.41, BDI z=2.46) 강세 지속으로 에너지 vs 비에너지 이분화 구조 심화. 달러 약세 가속(DXY z=-2.11)은 글로벌 유동성 완화 방향
PM 포지셔닝: 적극
내러티브: 오늘 시장의 핵심 사건은 WTI -7.3% 급락이다. 지정학 위험(GPR 304)이 여전히 극심한데 유가가 오히려 떨어졌다는 건, 시장이 공급 충격보다 수요 둔화를 더 무서워한다는 뜻이다. 디젤크랙(-12.5%)과 히팅오일(-9.3%) 동반 붕괴가 이를 확인해준다. 에너지는 아직 $91 이상이라 '확장기'이긴 하지만, 돈이 빠져나가기 시작한 섹터다.

반면 돈이 몰리는 곳은 분명하다. 반도체다. SOX 지수는 한 달에 +18% 올랐고, z-score 2.28로 과열이지만 멈출 기미가 없다. 오늘 ASML 실적, 내일 TSMC 실적이

## 활성 가설
- 가설: [S50_에너지]  교차 연결 — 크로스섹터 관점에서 '에너지 전환과 전력 인프라 투자 수혜 판단' 재해석 가능 (점수:17, 상태:proposed, 섹터:S50_에너지)
  내용: [S50_에너지]  교차 연결 — 크로스섹터 관점에서 '에너지 전환과 전력 인프라 투자 수혜 판단' 재해석 가능.
- 가설: [S10_반도체기술] '# 260321 rss 모음


## Lumotive makes optical breakthrough (점수:10, 상태:proposed, 섹터:S10_반도체기술)
  내용: [S10_반도체기술] '# 260321 rss 모음


## Lumotive makes optical breakthrough, targets 10,000-port da'의 시사점이 'AI반도체 수요 지속성 vs 피크아웃 리스크 판단'와 연결 가능 — 추가 검증 필요.
- 가설: **가설:**

만약 바이낸스의 규제 준수 강화 정책이 스테이블코인 자금 흐름을 기존 위협 자산에서、合コン플라이언스 에너지 인프라 프로젝트로 재배치한다면 (점수:21, 상태:proposed, 섹터:S50_에너지)
  내용: **가설:**

만약 바이낸스의 규제 준수 강화 정책이 스테이블코인 자금 흐름을 기존 위협 자산에서、合コン플라이언스 에너지 인프라 프로젝트로 재배치한다면, 에너지 전환 관련 중소형 기업들의 자금 조달 비용이 단기적으로 하락할 것이며, 이는 KB전우제 참여 기업의 실질 
- 가설: **가설: 만약 이란 전쟁 충돌로 IEA가 4억배럴 전략비축유 방출→유가 급락이 발생한다면 (점수:19, 상태:proposed, 섹터:S30_산업재방산)
  내용: **가설: 만약 이란 전쟁 충돌로 IEA가 4억배럴 전략비축유 방출→유가 급락이 발생한다면,美元 약세 pressão와 함께 방산 수출 확대 중인 한국 방산株의 외화매출 기준 내재가치가 오히려 희석될 것이다.**

근거: IEA 방출은 사상 최대 규모로 유가 약세 압박 
- 가설: **가설:**

만약 금리 인상과 보유세 상승이 **동시에 가속화**된다면 (점수:18, 상태:propose
…(축약, 원문은 morning_briefing dry-run/stdout 참조)</pre>

<b>stderr tail</b>
<pre>no oauth token found for github.com</pre>

weekly dry-run

[briefing_wrapper] kind=weekly chars=486 meta={"done": 18, "queued": 5, "avg_kpi": "N/A", "discoveries": 0}
<b>📅 주간 운영·시장 리포트 · Telegram</b>
<i>2026-04-08 ~ 2026-04-15 / source=weekly-telegram-wrapper</i>

<b>운영</b>
완료 작업 18건 / 현재 대기 5건 / 에이전트 평균 KPI N/A점

<b>시장</b>
S&amp;P500: 조회 실패('NoneType' object is not subscriptable) / NASDAQ: 조회 실패('NoneType' object is not subscriptable) / DOW: 조회 실패('NoneType' object is not subscriptable)

<b>파이프라인 발견</b>
이번 주 발견 0건

<b>유동성</b>
연준 순유동성 $+5.95T (전주 $-0.1B) / 국면: 유동성 축소 — 흡수 국면 / 단기 위험자산 비중 축소. 연준 정책 전환 시점 주시.

<b>최근 일간 리포트</b>
일간 리포트: 데일리 마켓 인텔리전스

실제 발송 시도

명령:

/bin/bash /Users/ron/.hermes/workspace/scripts/pipeline/morning_briefing_wrapper.sh --no-llm

결과:

[briefing_wrapper] kind=morning chars=2995 meta={"morning_briefing_ok": true, "stderr_tail": "no oauth token found for github.com"}
[SECTOR_TRACE] send_sector(report) caller=telegram_briefing_wrapper.py text='<b>🌅 아침 투자 브리핑 · Telegram</b>\n<i>2026-04-15 16:40 KST / source=morning_briefing.'
Telegram send failed after 3 retries: <urlopen error [Errno 8] nodename nor servname provided, or not known>
[briefing_wrapper] send_result={"ok": false, "message_ids": [], "error": "[telegram] message failed: Network request for 'sendMessage' failed!\nHttpError: Network request for 'sendMessage' failed!", "chat_id": -1003522748967, "topic_id": 8, "transport": "openclaw_cli_fallback", "openclaw": {"raw_stdout": "[telegram] autoSelectFamily=true (default-node22)\n[telegram] dnsResultOrder=ipv4first (default-node22)"}, "stderr_tail": "[telegram] message failed: Network request for 'sendMessage' failed!\nHttpError: Network request for 'sendMessage' failed!"}

판정: - wrapper는 report topic id 8까지 정확히 라우팅했다. - 직접 Telegram API는 DNS 실패. - OpenClaw CLI fallback도 gateway 내부 네트워크 요청 실패. - 따라서 실제 Telegram message_id는 미생성.

launchd 등록 시도

명령:

uid=$(id -u)
launchctl bootstrap gui/$uid ~/Library/LaunchAgents/com.openclaw.morning-briefing-telegram.plist
launchctl bootstrap gui/$uid ~/Library/LaunchAgents/com.openclaw.weekly-briefing-telegram.plist

결과:

Bootstrap failed: 5: Input/output error
Try re-running the command as root for richer errors.

판정: - 이전 launchd 부활 작업과 같은 제약이다. - plist 자체는 정상이나 현재 Codex 세션에서 launchd bootstrap이 제한된다. - 일반 터미널에서 재실행 필요.

기존 TTS 유지 증거

기존 TTS 스크립트와 plist는 이번 작업에서 수정하지 않았다.

Apr 15 16:08:58 2026 /Users/ron/.hermes/skills/smart-home/scripts/tts_morning.sh
Apr 15 16:08:58 2026 /Users/ron/.hermes/skills/smart-home/scripts/tts_weekly.sh
Apr 15 16:12:09 2026 /Users/ron/Library/LaunchAgents/com.openclaw.morning-routine.plist
Apr 15 16:12:09 2026 /Users/ron/Library/LaunchAgents/com.openclaw.weekly-report.plist

다음에 일반 터미널에서 할 일

uid=$(id -u)
launchctl bootstrap gui/$uid ~/Library/LaunchAgents/com.openclaw.morning-briefing-telegram.plist
launchctl bootstrap gui/$uid ~/Library/LaunchAgents/com.openclaw.weekly-briefing-telegram.plist
launchctl list | grep -E 'morning-briefing-telegram|weekly-briefing-telegram'

네트워크가 정상인 터미널에서 즉시 발송 확인:

/bin/bash ~/.hermes/workspace/scripts/pipeline/morning_briefing_wrapper.sh --no-llm

성공하면 출력의 send_result.message_ids에 message_id가 기록된다.

스크립트 전문

morning_briefing_wrapper.sh

#!/bin/bash
set -euo pipefail
export HOME="/Users/ron"
export PATH="/opt/homebrew/bin:/usr/local/bin:/usr/bin:/bin:/usr/sbin:/sbin"
cd "$HOME/.hermes/workspace"
exec /usr/bin/python3 "$HOME/.hermes/workspace/scripts/pipeline/telegram_briefing_wrapper.py" --kind morning "$@"

weekly_briefing_wrapper.sh

#!/bin/bash
set -euo pipefail
export HOME="/Users/ron"
export PATH="/opt/homebrew/bin:/usr/local/bin:/usr/bin:/bin:/usr/sbin:/sbin"
cd "$HOME/.hermes/workspace"
exec /usr/bin/python3 "$HOME/.hermes/workspace/scripts/pipeline/telegram_briefing_wrapper.py" --kind weekly "$@"

telegram_briefing_wrapper.py

#!/usr/bin/env python3
"""Telegram report-topic wrappers for morning and weekly briefings.

Keeps the Google Home TTS launchd jobs untouched. This wrapper builds a text
briefing and sends it to the Hermes sector-group `report` topic.
"""
from __future__ import annotations

import argparse
import html
import json
import os
import sqlite3
import subprocess
import sys
import time
from datetime import date, datetime, timedelta, timezone
from pathlib import Path
from typing import Any

SCRIPT_DIR = Path(__file__).resolve().parent
SCRIPTS_DIR = SCRIPT_DIR.parent
WORKSPACE = Path.home() / ".hermes" / "workspace"
MEMORY = WORKSPACE / "memory"
LOG_DIR = Path.home() / ".hermes" / "logs"
DB_PATH = Path.home() / ".hermes" / "data" / "ops_multiagent.db"
KST = timezone(timedelta(hours=9))

sys.path.insert(0, str(SCRIPTS_DIR))


def _now() -> datetime:
    return datetime.now(KST)


def _read_json(path: Path, default: Any = None) -> Any:
    try:
        if path.exists():
            return json.loads(path.read_text(encoding="utf-8"))
    except Exception:
        pass
    return default


def _escape(text: Any) -> str:
    return html.escape(str(text or ""), quote=False)


def _strip_html(text: str) -> str:
    # Telegram text body still benefits from escaped HTML; this is only for previews.
    return str(text or "").replace("\x00", "").strip()


def _fmt_t(val: Any) -> str:
    try:
        val = float(val)
    except Exception:
        return "N/A"
    sign = "+" if val > 0 else ""
    if abs(val) < 0.01:
        return f"${sign}{val * 1000:.1f}B"
    return f"${sign}{val:.2f}T"


def _fed_liquidity_oneliner() -> str:
    fl = _read_json(MEMORY / "fed-liquidity" / "latest.json", {}) or {}
    if not isinstance(fl, dict) or not fl:
        return "연준 유동성: 데이터 없음"
    net = _fmt_t(fl.get("net_t"))
    wow = _fmt_t(fl.get("wow_t"))
    phase = fl.get("phase") or "?"
    signal = fl.get("investment_signal") or fl.get("interpretation") or ""
    tail = f" / {signal}" if signal else ""
    return f"연준 순유동성 {net} (전주 {wow}) / 국면: {phase}{tail}"


def _bond_oneliner() -> str:
    data = _read_json(MEMORY / "bond-briefing" / "latest.json", {}) or {}
    if not isinstance(data, dict) or not data:
        return "채권: latest.json 없음"
    candidates = [
        data.get("summary"), data.get("headline"), data.get("narrative"),
        data.get("market_summary"), data.get("commentary"),
    ]
    for c in candidates:
        if isinstance(c, str) and c.strip():
            return "채권: " + c.strip()[:260]
    # Fall back to compact key metrics.
    bits = []
    for key in ("date", "us10y", "us2y", "kr3y", "curve", "signal"):
        if key in data:
            bits.append(f"{key}={data.get(key)}")
    return "채권: " + (", ".join(bits[:6]) if bits else "요약 필드 없음")


def _daily_report_oneliner() -> str:
    latest = MEMORY / "daily-report" / "report-2026-04-14.md"
    reports = sorted((MEMORY / "daily-report").glob("report-*.md"))
    if reports:
        latest = reports[-1]
    try:
        text = latest.read_text(encoding="utf-8")
    except Exception:
        return "일간 리포트: 없음"
    lines = [ln.strip("# *-\t ") for ln in text.splitlines() if ln.strip()]
    return "일간 리포트: " + (lines[0][:220] if lines else latest.name)


def _event_oneliner() -> str:
    data = _read_json(MEMORY / "economic-calendar" / "latest.json", {}) or {}
    events = []
    if isinstance(data, dict):
        raw = data.get("events") or data.get("data") or []
        if isinstance(raw, list):
            for ev in raw[:5]:
                if isinstance(ev, dict):
                    title = ev.get("event") or ev.get("title") or ev.get("name")
                    when = ev.get("time") or ev.get("hour") or ""
                    if title:
                        events.append(f"{when} {title}".strip())
                elif isinstance(ev, str):
                    events.append(ev[:120])
    return "오늘 일정: " + (" / ".join(events[:5]) if events else "캘린더 데이터 없음")


def _run_morning_briefing(use_llm: bool = True, timeout: int = 240) -> tuple[bool, str, str]:
    cmd = [sys.executable, str(SCRIPT_DIR / "morning_briefing.py"), "--mode", "morning", "--dry-run"]
    if not use_llm:
        cmd.append("--no-llm")
    env = os.environ.copy()
    env.setdefault("HOME", str(Path.home()))
    try:
        proc = subprocess.run(
            cmd,
            cwd=str(WORKSPACE),
            env=env,
            text=True,
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
            timeout=timeout,
        )
        out = (proc.stdout or "").strip()
        err = (proc.stderr or "").strip()
        return proc.returncode == 0 and bool(out), out, err
    except subprocess.TimeoutExpired as e:
        return False, (e.stdout or "") if isinstance(e.stdout, str) else "", f"timeout after {timeout}s"
    except Exception as e:
        return False, "", repr(e)


def _truncate_preserving(text: str, max_len: int = 2500) -> str:
    text = _strip_html(text)
    if len(text) <= max_len:
        return text
    return text[:max_len].rstrip() + "\n…(축약, 원문은 morning_briefing dry-run/stdout 참조)"


def build_morning_message(use_llm: bool = True) -> tuple[str, dict]:
    ok, body, err = _run_morning_briefing(use_llm=use_llm)
    meta = {"morning_briefing_ok": ok, "stderr_tail": err[-1200:] if err else ""}
    if not body:
        body = "morning_briefing.py 출력 없음 — 핵심 데이터만 전송"

    now = _now()
    llm_label = "LLM" if use_llm else "no-LLM"
    parts = [
        f"<b>🌅 아침 투자 브리핑 · Telegram</b>",
        f"<i>{_escape(now.strftime('%Y-%m-%d %H:%M KST'))} / source=morning_briefing.py({llm_label})</i>",
        "",
        f"<b>유동성</b>\n{_escape(_fed_liquidity_oneliner())}",
        "",
        f"<b>채권</b>\n{_escape(_bond_oneliner())}",
        "",
        f"<b>오늘 일정</b>\n{_escape(_event_oneliner())}",
        "",
        f"<b>브리핑 본문</b>\n<pre>{_escape(_truncate_preserving(body, 2600))}</pre>",
    ]
    if err:
        parts.append(f"\n<b>stderr tail</b>\n<pre>{_escape(err[-700:])}</pre>")
    return "\n".join(parts), meta


def _db_scalar(sql: str, params: tuple = (), default: Any = None) -> Any:
    try:
        if not DB_PATH.exists():
            return default
        with sqlite3.connect(str(DB_PATH), timeout=5) as db:
            return db.execute(sql, params).fetchone()[0]
    except Exception:
        return default


def _weekly_market_lines() -> list[str]:
    try:
        import yfinance as yf  # type: ignore
    except Exception as e:
        return [f"시장: yfinance import 실패({e})"]
    lines = []
    for kr_name, ticker in {"S&P500": "^GSPC", "NASDAQ": "^IXIC", "DOW": "^DJI"}.items():
        try:
            h = yf.Ticker(ticker).history(period="5d")
            if len(h) >= 2:
                first, last = h["Close"].iloc[0], h["Close"].iloc[-1]
                chg = (last - first) / first * 100
                lines.append(f"{kr_name}: {chg:+.1f}%")
        except Exception as e:
            lines.append(f"{kr_name}: 조회 실패({str(e)[:60]})")
    return lines or ["시장: 데이터 없음"]


def _weekly_discovery_count(today: date) -> int:
    ideas_dir = WORKSPACE / "reports" / "ideas"
    count = 0
    for i in range(7):
        ds = (today - timedelta(days=i)).strftime("%Y%m%d")
        path = ideas_dir / f"scored_discoveries_{ds}.jsonl"
        try:
            if path.exists():
                with path.open(encoding="utf-8") as fh:
                    count += sum(1 for _ in fh)
        except Exception:
            pass
    return count


def build_weekly_message() -> tuple[str, dict]:
    today = _now().date()
    week_ago = today - timedelta(days=7)
    done = _db_scalar(
        "SELECT COUNT(*) FROM bus_commands WHERE status='done' AND date(updated_at) >= ?",
        (week_ago.isoformat(),),
        0,
    )
    queued = _db_scalar("SELECT COUNT(*) FROM bus_commands WHERE status='queued'", default=0)
    avg_kpi = _db_scalar(
        "SELECT AVG(score) FROM agent_kpi_daily WHERE date >= ? AND score IS NOT NULL",
        (week_ago.isoformat(),),
        None,
    )
    avg_txt = f"{float(avg_kpi):.1f}" if avg_kpi is not None else "N/A"
    discoveries = _weekly_discovery_count(today)
    market = " / ".join(_weekly_market_lines())
    message = "\n".join([
        "<b>📅 주간 운영·시장 리포트 · Telegram</b>",
        f"<i>{_escape(week_ago.isoformat())} ~ {_escape(today.isoformat())} / source=weekly-telegram-wrapper</i>",
        "",
        f"<b>운영</b>\n완료 작업 {done}건 / 현재 대기 {queued}건 / 에이전트 평균 KPI {avg_txt}점",
        "",
        f"<b>시장</b>\n{_escape(market)}",
        "",
        f"<b>파이프라인 발견</b>\n이번 주 발견 {discoveries}건",
        "",
        f"<b>유동성</b>\n{_escape(_fed_liquidity_oneliner())}",
        "",
        f"<b>최근 일간 리포트</b>\n{_escape(_daily_report_oneliner())}",
    ])
    return message, {"done": done, "queued": queued, "avg_kpi": avg_txt, "discoveries": discoveries}


def _send_report_topic(message: str, parse_mode: str = "HTML") -> dict:
    from shared import telegram

    chat_id, topics = telegram._load_sector_config()  # same config as send_sector('report')
    topic_id = (topics or {}).get("report")
    trace = f"[SECTOR_TRACE] send_sector(report) caller=telegram_briefing_wrapper.py text={message[:80]!r}"
    print(trace, flush=True)
    try:
        LOG_DIR.mkdir(parents=True, exist_ok=True)
        with (LOG_DIR / "sector_trace.log").open("a", encoding="utf-8") as tf:
            tf.write(f"{datetime.now().strftime('%Y-%m-%d %H:%M:%S')} {trace}\n")
    except Exception:
        pass

    if chat_id is None or topic_id is None:
        return {"ok": False, "message_ids": [], "error": "missing sector report config"}

    chunks = telegram.split_message(message)
    # Do not pre-write to telegram_dedup here. shared.telegram._is_duplicate records
    # before network send, so a DNS/API failure would poison the cache and make
    # the next real launchd run look like a successful duplicate skip. These
    # launchd wrappers run at most once per schedule, so raw send is safer.
    mids: list[int] = []
    ok_all = True
    last_error = None
    for i, chunk in enumerate(chunks):
        result = telegram._send_raw(chat_id, chunk, parse_mode=parse_mode, topic_id=topic_id)
        if result.get("ok"):
            mid = result.get("message_id")
            if mid is not None:
                mids.append(int(mid))
        else:
            ok_all = False
            last_error = result.get("error")
        if i < len(chunks) - 1:
            time.sleep(0.5)

    if ok_all:
        return {"ok": True, "message_ids": mids, "error": None, "chat_id": chat_id, "topic_id": topic_id, "transport": "telegram_api"}

    # Fallback: ask the already-running OpenClaw/Hermes gateway to send. This
    # works in Codex sessions where direct DNS to api.telegram.org is blocked.
    try:
        cli = subprocess.run(
            [
                "openclaw", "message", "send",
                "--channel", "telegram",
                "--target", str(chat_id),
                "--thread-id", str(topic_id),
                "--message", message,
                "--json",
            ],
            cwd=str(WORKSPACE),
            text=True,
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
            timeout=45,
        )
        cli_out = (cli.stdout or "").strip()
        cli_err = (cli.stderr or "").strip()
        parsed = {}
        try:
            parsed = json.loads(cli_out) if cli_out else {}
        except Exception:
            parsed = {"raw_stdout": cli_out}
        cli_ok = cli.returncode == 0 and (not parsed or parsed.get("ok", True) is not False)
        return {
            "ok": cli_ok,
            "message_ids": mids,
            "error": None if cli_ok else (cli_err or last_error or cli_out),
            "chat_id": chat_id,
            "topic_id": topic_id,
            "transport": "openclaw_cli_fallback",
            "openclaw": parsed or cli_out[-500:],
            "stderr_tail": cli_err[-500:],
        }
    except Exception as e:
        return {"ok": False, "message_ids": mids, "error": f"{last_error}; fallback_failed={e!r}", "chat_id": chat_id, "topic_id": topic_id, "transport": "failed"}


def main() -> int:
    parser = argparse.ArgumentParser(description="Send morning/weekly briefing to Telegram report topic")
    parser.add_argument("--kind", choices=["morning", "weekly"], required=True)
    parser.add_argument("--no-llm", action="store_true", help="morning only: call morning_briefing.py --no-llm")
    parser.add_argument("--dry-run", action="store_true", help="build message only, do not send")
    args = parser.parse_args()

    if args.kind == "morning":
        message, meta = build_morning_message(use_llm=not args.no_llm)
    else:
        message, meta = build_weekly_message()

    print(f"[briefing_wrapper] kind={args.kind} chars={len(message)} meta={json.dumps(meta, ensure_ascii=False)}")
    if args.dry_run:
        print(message)
        return 0

    result = _send_report_topic(message, parse_mode="HTML")
    print(f"[briefing_wrapper] send_result={json.dumps(result, ensure_ascii=False)}")
    return 0 if result.get("ok") else 1


if __name__ == "__main__":
    raise SystemExit(main())

자체평가

정확성: 4.2/5 — TTS 유지 + Telegram wrapper/plist 설계와 구현은 완료. 단, 실제 Telegram 발송은 네트워크 제한으로 실패.
완성도: 4.4/5 — morning/weekly 모두 wrapper와 plist 있음. message_id는 네트워크 정상 환경에서만 확보 가능.
검증: 3.8/5 — py_compile, bash -n, plutil, dry-run, 실제 발송 시도까지 했으나 외부 네트워크/launchctl은 세션 제한.
최소 변경: 4.8/5 — 기존 TTS/원본 morning_briefing.py 미수정, 신규 파일만 추가.

종합: 4.3/5

Remaining Risks: - 현재 Codex 세션의 DNS/Telegram API 제한 때문에 message_id까지 확인하지 못했다. - 현재 Codex 세션의 launchctl bootstrap 제한 때문에 신규 plist는 파일만 생성됐고 로드 상태는 아니다. - scheduled 실행은 기본적으로 LLM 사용 모드다. LLM 장애 시 기존 morning_briefing.py의 fallback 또는 wrapper의 stderr tail로 관측해야 한다.

morning-routine / weekly-report Telegram wrapper 설계·구현

결론