772 lines
27 KiB
Python
772 lines
27 KiB
Python
"""
|
||
Channel Hub — AwoooP 入站事件統一路由 + Progressive Feedback Policy
|
||
====================================================================
|
||
AwoooP Phase 7: ADR-106（channel_event family）
|
||
2026-05-04 ogt + Claude Sonnet 4.6
|
||
|
||
功能：
|
||
1. Telegram 入站事件鏡像（記錄到 awooop_conversation_event）
|
||
2. 建立 platform run（呼叫 platform_runtime.create_run）
|
||
3. Progressive Feedback Policy：
|
||
- run 進入 WAITING_TOOL 狀態 → 30 秒後若未 complete → 發 interim Telegram 訊息
|
||
- 訊息記錄到 awooop_outbound_message
|
||
4. Shadow Mode：不發任何 Telegram 訊息（只記錄到 outbound_message, status='shadow'）
|
||
|
||
Progressive Feedback Policy 設計（ADR-106 P2-03）：
|
||
- 用 asyncio.create_task 啟動 30s 計時器
|
||
- 30s 後查詢 run state：若仍在 WAITING_TOOL → 發 interim 訊息
|
||
- interim 訊息：「AI 正在分析中，請稍候...」（不洩漏 run 細節）
|
||
- Final reply 由 shadow_execute() 完成後觸發（Phase 8 實作）
|
||
|
||
與 legacy telegram_gateway.py 的關係：
|
||
- 完全獨立，不修改 legacy gateway
|
||
- legacy 繼續處理 legacy flow（signal_worker 觸發的 approval/notification）
|
||
- AwoooP run 只走本模組
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import asyncio
|
||
import hashlib
|
||
import html
|
||
import json
|
||
from datetime import UTC, datetime
|
||
from typing import Any
|
||
from uuid import NAMESPACE_URL, UUID, uuid5
|
||
|
||
import structlog
|
||
from sqlalchemy import select, text
|
||
from sqlalchemy.ext.asyncio import AsyncSession
|
||
|
||
from src.db.awooop_models import AwoooPRunState
|
||
from src.services.audit_sink import _redact_string, sanitize
|
||
from src.services.platform_runtime import create_run
|
||
|
||
logger = structlog.get_logger(__name__)
|
||
|
||
# Progressive Feedback Policy：等待超過此秒數才發 interim 訊息
|
||
_INTERIM_WAIT_SECONDS = 30
|
||
_OUTBOUND_REDACTION_VERSION = "audit_sink_v1"
|
||
|
||
|
||
def _input_sha256(input_payload: dict[str, Any] | None) -> str | None:
|
||
"""計算 Run input 的穩定 hash，讓 mirror run 也能保留最小完整性證據。"""
|
||
if not input_payload:
|
||
return None
|
||
canonical = json.dumps(
|
||
input_payload,
|
||
sort_keys=True,
|
||
separators=(",", ":"),
|
||
ensure_ascii=False,
|
||
)
|
||
return hashlib.sha256(canonical.encode()).hexdigest()
|
||
|
||
|
||
async def ensure_completed_shadow_run(
|
||
db: AsyncSession,
|
||
*,
|
||
project_id: str,
|
||
run_id: UUID,
|
||
agent_id: str,
|
||
trigger_type: str,
|
||
trigger_ref: str | None,
|
||
input_payload: dict[str, Any] | None = None,
|
||
) -> bool:
|
||
"""為 legacy mirror 資料補一筆 completed shadow run。
|
||
|
||
AwoooP 在 strangler 階段會先 mirror legacy Telegram / alert-grouping
|
||
資料。這些事件不應重新觸發 runtime，但需要 run_state 當 Console 的
|
||
聚合錨點；因此這裡建立的是已完成的 shadow run，不會被 worker pick up。
|
||
"""
|
||
result = await db.execute(
|
||
text("""
|
||
INSERT INTO awooop_run_state (
|
||
run_id, project_id, agent_id, state,
|
||
trigger_type, trigger_ref, is_shadow,
|
||
input_sha256,
|
||
attempt_count, max_attempts, cost_usd, step_count,
|
||
created_at, completed_at, timeout_at
|
||
) VALUES (
|
||
:run_id, :project_id, :agent_id, 'completed',
|
||
:trigger_type, :trigger_ref, TRUE,
|
||
:input_sha256,
|
||
0, 3, 0.0000, 0,
|
||
NOW(), NOW(), NOW()
|
||
)
|
||
ON CONFLICT (run_id) DO NOTHING
|
||
RETURNING run_id
|
||
"""),
|
||
{
|
||
"run_id": run_id,
|
||
"project_id": project_id,
|
||
"agent_id": agent_id,
|
||
"trigger_type": trigger_type,
|
||
"trigger_ref": trigger_ref,
|
||
"input_sha256": _input_sha256(input_payload),
|
||
},
|
||
)
|
||
inserted = result.fetchone() is not None
|
||
if inserted:
|
||
logger.info(
|
||
"completed_shadow_run_created",
|
||
project_id=project_id,
|
||
run_id=str(run_id),
|
||
agent_id=agent_id,
|
||
trigger_type=trigger_type,
|
||
)
|
||
return inserted
|
||
|
||
|
||
def build_grouped_alert_run_id(project_id: str, provider_event_id: str) -> UUID:
|
||
"""為 grouped child alert 建立穩定 run_id，讓 Run Monitor 可回查。"""
|
||
return uuid5(NAMESPACE_URL, f"awooop:grouped-alert:{project_id}:{provider_event_id}")
|
||
|
||
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
# 入站事件記錄
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
|
||
async def mirror_inbound_event(
|
||
db: AsyncSession,
|
||
*,
|
||
project_id: str,
|
||
channel_type: str,
|
||
provider_event_id: str,
|
||
platform_subject_id: str | None = None,
|
||
channel_user_id: str | None = None,
|
||
channel_chat_id: str | None = None,
|
||
content_type: str = "text",
|
||
raw_content: str | None = None,
|
||
attachment_sha256: str | None = None,
|
||
provider_ts: datetime | None = None,
|
||
run_id: UUID | None = None,
|
||
is_duplicate: bool = False,
|
||
) -> UUID:
|
||
"""
|
||
記錄入站 channel event 到 awooop_conversation_event。
|
||
|
||
raw_content 只用於計算 hash 和 preview，不入庫明文。
|
||
回傳 event_id。
|
||
"""
|
||
content_hash: str | None = None
|
||
content_preview: str | None = None
|
||
|
||
if raw_content is not None:
|
||
content_hash = hashlib.sha256(raw_content.encode()).hexdigest()
|
||
# preview：redact 後截取前 256 字元
|
||
redacted = _redact_string(raw_content)
|
||
content_preview = redacted[:256] if len(redacted) > 256 else redacted
|
||
|
||
result = await db.execute(
|
||
text("""
|
||
INSERT INTO awooop_conversation_event (
|
||
project_id, channel_type, provider_event_id,
|
||
platform_subject_id, channel_user_id, channel_chat_id,
|
||
run_id, content_type, content_hash, content_preview,
|
||
attachment_sha256, is_duplicate, provider_ts, received_at
|
||
) VALUES (
|
||
:project_id, :channel_type, :provider_event_id,
|
||
:platform_subject_id, :channel_user_id, :channel_chat_id,
|
||
:run_id, :content_type, :content_hash, :content_preview,
|
||
:attachment_sha256, :is_duplicate, :provider_ts, NOW()
|
||
)
|
||
ON CONFLICT (project_id, channel_type, provider_event_id) DO UPDATE SET
|
||
is_duplicate = TRUE,
|
||
run_id = COALESCE(EXCLUDED.run_id, awooop_conversation_event.run_id)
|
||
RETURNING event_id
|
||
"""),
|
||
{
|
||
"project_id": project_id,
|
||
"channel_type": channel_type,
|
||
"provider_event_id": provider_event_id,
|
||
"platform_subject_id": platform_subject_id,
|
||
"channel_user_id": channel_user_id,
|
||
"channel_chat_id": channel_chat_id,
|
||
"run_id": run_id,
|
||
"content_type": content_type,
|
||
"content_hash": content_hash,
|
||
"content_preview": content_preview,
|
||
"attachment_sha256": attachment_sha256,
|
||
"is_duplicate": is_duplicate,
|
||
"provider_ts": provider_ts,
|
||
},
|
||
)
|
||
row = result.fetchone()
|
||
event_id: UUID = row[0]
|
||
logger.info(
|
||
"channel_event_mirrored",
|
||
project_id=project_id,
|
||
channel_type=channel_type,
|
||
event_id=str(event_id),
|
||
is_duplicate=is_duplicate,
|
||
)
|
||
return event_id
|
||
|
||
|
||
def build_grouped_alert_provider_event_id(alert_id: str, fingerprint: str) -> str:
|
||
"""建立 grouped child alert 的冪等 provider_event_id。"""
|
||
safe_alert_id = str(alert_id).strip() or "unknown"
|
||
safe_fingerprint = str(fingerprint).strip()[:32] or "no-fingerprint"
|
||
return f"alert-group:{safe_alert_id}:{safe_fingerprint}"
|
||
|
||
|
||
def format_grouped_alert_event_content(
|
||
*,
|
||
alert_id: str,
|
||
alertname: str,
|
||
severity: str,
|
||
namespace: str,
|
||
target_resource: str,
|
||
group_key: str,
|
||
count: int,
|
||
parent_fingerprint: str | None,
|
||
fingerprint: str,
|
||
) -> str:
|
||
"""格式化只落 AwoooP、不發 Telegram 的告警收斂事件摘要。"""
|
||
parent = parent_fingerprint or "-"
|
||
target = target_resource or "-"
|
||
ns = namespace or "default"
|
||
return "\n".join(
|
||
[
|
||
"告警已收斂，不發 Telegram",
|
||
f"Alert ID: {alert_id}",
|
||
f"Alert: {alertname}",
|
||
f"Severity: {severity}",
|
||
f"Namespace: {ns}",
|
||
f"Target: {target}",
|
||
f"Group: {group_key}",
|
||
f"Group Count: {count}",
|
||
f"Parent Fingerprint: {parent}",
|
||
f"Child Fingerprint: {fingerprint}",
|
||
]
|
||
)
|
||
|
||
|
||
def format_grouped_alert_digest_text(
|
||
*,
|
||
alertname: str,
|
||
severity: str,
|
||
namespace: str,
|
||
target_resource: str,
|
||
group_key: str,
|
||
count: int,
|
||
) -> str:
|
||
"""格式化要回覆到父告警卡的短 digest。"""
|
||
safe_alert = html.escape(alertname or "unknown")
|
||
safe_severity = html.escape(severity or "unknown")
|
||
safe_namespace = html.escape(namespace or "default")
|
||
safe_target = html.escape(target_resource or "unknown")
|
||
safe_group = html.escape(group_key or "unknown")
|
||
|
||
return "\n".join(
|
||
[
|
||
"🧩 <b>告警已收斂到父卡</b>",
|
||
f"├ 類型：<code>{safe_alert}</code>",
|
||
f"├ 等級：<code>{safe_severity}</code>",
|
||
f"├ 範圍：<code>{safe_namespace}</code>",
|
||
f"├ 最新目標：<code>{safe_target}</code>",
|
||
f"├ 群組：<code>{safe_group}</code>",
|
||
f"└ 目前視窗：<b>{count}</b> 筆同組告警",
|
||
"",
|
||
"完整子告警請看 AwoooP Run 監控，不再逐筆發 Telegram。",
|
||
]
|
||
)
|
||
|
||
|
||
async def maybe_send_grouped_alert_digest(
|
||
*,
|
||
project_id: str,
|
||
alertname: str,
|
||
severity: str,
|
||
namespace: str,
|
||
target_resource: str,
|
||
group_key: str,
|
||
count: int,
|
||
parent_fingerprint: str | None,
|
||
) -> bool:
|
||
"""若父告警卡已存在，回覆一則低頻 digest；找不到父卡則安靜降級。"""
|
||
if not parent_fingerprint:
|
||
return False
|
||
|
||
try:
|
||
from sqlalchemy import select
|
||
|
||
from src.db.base import get_db_context
|
||
from src.db.models import ApprovalRecord
|
||
from src.services.telegram_gateway import get_telegram_gateway
|
||
|
||
async with get_db_context(project_id) as db:
|
||
result = await db.execute(
|
||
select(ApprovalRecord.incident_id)
|
||
.where(ApprovalRecord.fingerprint == parent_fingerprint)
|
||
.where(ApprovalRecord.incident_id.is_not(None))
|
||
.order_by(ApprovalRecord.created_at.desc())
|
||
.limit(1)
|
||
)
|
||
incident_id = result.scalar_one_or_none()
|
||
|
||
if not incident_id:
|
||
logger.info(
|
||
"grouped_alert_digest_parent_not_ready",
|
||
project_id=project_id,
|
||
group_key=group_key,
|
||
parent_fingerprint=parent_fingerprint,
|
||
)
|
||
return False
|
||
|
||
digest_text = format_grouped_alert_digest_text(
|
||
alertname=alertname,
|
||
severity=severity,
|
||
namespace=namespace,
|
||
target_resource=target_resource,
|
||
group_key=group_key,
|
||
count=count,
|
||
)
|
||
sent = await get_telegram_gateway().append_grouped_alert_digest(
|
||
incident_id=str(incident_id),
|
||
group_key=group_key,
|
||
digest_text=digest_text,
|
||
)
|
||
logger.info(
|
||
"grouped_alert_digest_result",
|
||
project_id=project_id,
|
||
incident_id=str(incident_id),
|
||
group_key=group_key,
|
||
count=count,
|
||
sent=sent,
|
||
)
|
||
return sent
|
||
except Exception as exc:
|
||
logger.warning(
|
||
"grouped_alert_digest_failed",
|
||
project_id=project_id,
|
||
group_key=group_key,
|
||
parent_fingerprint=parent_fingerprint,
|
||
error=str(exc),
|
||
)
|
||
return False
|
||
|
||
|
||
async def record_grouped_alert_event(
|
||
*,
|
||
project_id: str,
|
||
alert_id: str,
|
||
alertname: str,
|
||
severity: str,
|
||
namespace: str,
|
||
target_resource: str,
|
||
group_key: str,
|
||
count: int,
|
||
parent_fingerprint: str | None,
|
||
fingerprint: str,
|
||
) -> UUID | None:
|
||
"""
|
||
將被 AlertGroupingService 收斂的子告警落到 AwoooP conversation_event。
|
||
|
||
這條路徑刻意不發 Telegram，只保留 operator-facing 脈絡：
|
||
- 群組不洗版
|
||
- Console 仍能看到同組告警正在持續發生
|
||
- DB 失敗 fail-open，不影響 Alertmanager webhook ACK
|
||
"""
|
||
try:
|
||
from src.db.base import get_db_context
|
||
|
||
provider_event_id = build_grouped_alert_provider_event_id(alert_id, fingerprint)
|
||
content = format_grouped_alert_event_content(
|
||
alert_id=alert_id,
|
||
alertname=alertname,
|
||
severity=severity,
|
||
namespace=namespace,
|
||
target_resource=target_resource,
|
||
group_key=group_key,
|
||
count=count,
|
||
parent_fingerprint=parent_fingerprint,
|
||
fingerprint=fingerprint,
|
||
)
|
||
|
||
async with get_db_context(project_id) as db:
|
||
run_id = build_grouped_alert_run_id(project_id, provider_event_id)
|
||
await ensure_completed_shadow_run(
|
||
db,
|
||
project_id=project_id,
|
||
run_id=run_id,
|
||
agent_id="legacy-alert-grouping",
|
||
trigger_type="grouped_alert_event",
|
||
trigger_ref=provider_event_id,
|
||
input_payload={
|
||
"alert_id": alert_id,
|
||
"alertname": alertname,
|
||
"severity": severity,
|
||
"group_key": group_key,
|
||
"fingerprint": fingerprint,
|
||
},
|
||
)
|
||
event_id = await mirror_inbound_event(
|
||
db,
|
||
project_id=project_id,
|
||
channel_type="internal",
|
||
provider_event_id=provider_event_id,
|
||
platform_subject_id="alertmanager",
|
||
channel_user_id="alertmanager",
|
||
channel_chat_id=f"alert-group:{group_key}",
|
||
content_type="text",
|
||
raw_content=content,
|
||
provider_ts=datetime.now(UTC),
|
||
run_id=run_id,
|
||
)
|
||
|
||
logger.info(
|
||
"grouped_alert_event_recorded",
|
||
project_id=project_id,
|
||
alert_id=alert_id,
|
||
event_id=str(event_id),
|
||
group_key=group_key,
|
||
count=count,
|
||
)
|
||
await maybe_send_grouped_alert_digest(
|
||
project_id=project_id,
|
||
alertname=alertname,
|
||
severity=severity,
|
||
namespace=namespace,
|
||
target_resource=target_resource,
|
||
group_key=group_key,
|
||
count=count,
|
||
parent_fingerprint=parent_fingerprint,
|
||
)
|
||
return event_id
|
||
except Exception as exc:
|
||
logger.warning(
|
||
"grouped_alert_event_record_failed",
|
||
project_id=project_id,
|
||
alert_id=alert_id,
|
||
group_key=group_key,
|
||
error=str(exc),
|
||
)
|
||
return None
|
||
|
||
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
# 出站訊息記錄
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
|
||
async def record_outbound_message(
|
||
db: AsyncSession,
|
||
*,
|
||
project_id: str,
|
||
run_id: UUID,
|
||
channel_type: str,
|
||
channel_chat_id: str,
|
||
message_type: str, # 'interim' | 'final' | 'error' | 'approval_request'
|
||
content: str | None = None,
|
||
source_envelope: dict[str, Any] | None = None,
|
||
provider_message_id: str | None = None,
|
||
send_status: str = "pending",
|
||
conversation_event_id: UUID | None = None,
|
||
triggered_by_state: str | None = None,
|
||
waiting_since: datetime | None = None,
|
||
is_shadow: bool = True,
|
||
) -> UUID:
|
||
"""
|
||
記錄出站訊息到 awooop_outbound_message。
|
||
|
||
is_shadow=True：status='shadow'（不實際發送，只記錄）
|
||
"""
|
||
content_hash: str | None = None
|
||
content_preview: str | None = None
|
||
content_redacted: str | None = None
|
||
if content is not None:
|
||
content_hash = hashlib.sha256(content.encode()).hexdigest()
|
||
content_redacted = _redact_string(content)
|
||
content_preview = content_redacted[:256]
|
||
|
||
envelope: dict[str, Any] = sanitize(source_envelope or {})
|
||
envelope.update({
|
||
"schema_version": "outbound_source_envelope_v1",
|
||
"redaction_version": _OUTBOUND_REDACTION_VERSION,
|
||
"content_sha256": content_hash,
|
||
"content_length": len(content) if content is not None else 0,
|
||
})
|
||
source_envelope_json = json.dumps(envelope, ensure_ascii=False, default=str)
|
||
|
||
actual_status = "shadow" if is_shadow else send_status
|
||
sent_at = datetime.now(UTC) if actual_status == "sent" else None
|
||
|
||
await ensure_completed_shadow_run(
|
||
db,
|
||
project_id=project_id,
|
||
run_id=run_id,
|
||
agent_id="legacy-telegram-gateway",
|
||
trigger_type="legacy_outbound",
|
||
trigger_ref=provider_message_id,
|
||
input_payload={
|
||
"channel_type": channel_type,
|
||
"channel_chat_id": channel_chat_id,
|
||
"message_type": message_type,
|
||
"send_status": actual_status,
|
||
"triggered_by_state": triggered_by_state,
|
||
},
|
||
)
|
||
|
||
result = await db.execute(
|
||
text("""
|
||
INSERT INTO awooop_outbound_message (
|
||
project_id, run_id, conversation_event_id,
|
||
channel_type, channel_chat_id, message_type,
|
||
content_hash, content_preview, content_redacted,
|
||
redaction_version, source_envelope,
|
||
provider_message_id,
|
||
send_status, queued_at, sent_at,
|
||
triggered_by_state, waiting_since
|
||
) VALUES (
|
||
:project_id, :run_id, :conversation_event_id,
|
||
:channel_type, :channel_chat_id, :message_type,
|
||
:content_hash, :content_preview, :content_redacted,
|
||
:redaction_version, CAST(:source_envelope AS jsonb),
|
||
:provider_message_id,
|
||
:send_status, NOW(), :sent_at,
|
||
:triggered_by_state, :waiting_since
|
||
)
|
||
RETURNING message_id
|
||
"""),
|
||
{
|
||
"project_id": project_id,
|
||
"run_id": run_id,
|
||
"conversation_event_id": conversation_event_id,
|
||
"channel_type": channel_type,
|
||
"channel_chat_id": channel_chat_id,
|
||
"message_type": message_type,
|
||
"content_hash": content_hash,
|
||
"content_preview": content_preview,
|
||
"content_redacted": content_redacted,
|
||
"redaction_version": _OUTBOUND_REDACTION_VERSION,
|
||
"source_envelope": source_envelope_json,
|
||
"provider_message_id": provider_message_id,
|
||
"send_status": actual_status,
|
||
"sent_at": sent_at,
|
||
"triggered_by_state": triggered_by_state,
|
||
"waiting_since": waiting_since,
|
||
},
|
||
)
|
||
row = result.fetchone()
|
||
message_id: UUID = row[0]
|
||
logger.info(
|
||
"outbound_message_recorded",
|
||
project_id=project_id,
|
||
run_id=str(run_id),
|
||
message_type=message_type,
|
||
send_status=actual_status,
|
||
message_id=str(message_id),
|
||
)
|
||
return message_id
|
||
|
||
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
# Progressive Feedback Policy
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
|
||
async def schedule_interim_feedback(
|
||
*,
|
||
project_id: str,
|
||
run_id: UUID,
|
||
channel_type: str,
|
||
channel_chat_id: str,
|
||
conversation_event_id: UUID | None = None,
|
||
is_shadow: bool = True,
|
||
wait_seconds: int = _INTERIM_WAIT_SECONDS,
|
||
) -> None:
|
||
"""
|
||
Progressive Feedback Policy：
|
||
等待 wait_seconds 秒後，若 run 仍在 WAITING_TOOL → 發 interim 訊息。
|
||
|
||
Shadow Mode：記錄到 outbound_message(status='shadow')，不實際發 Telegram 訊息。
|
||
"""
|
||
asyncio.create_task(
|
||
_interim_feedback_task(
|
||
project_id=project_id,
|
||
run_id=run_id,
|
||
channel_type=channel_type,
|
||
channel_chat_id=channel_chat_id,
|
||
conversation_event_id=conversation_event_id,
|
||
is_shadow=is_shadow,
|
||
wait_seconds=wait_seconds,
|
||
),
|
||
name=f"interim_feedback_{str(run_id)[:8]}",
|
||
)
|
||
|
||
|
||
async def _interim_feedback_task(
|
||
*,
|
||
project_id: str,
|
||
run_id: UUID,
|
||
channel_type: str,
|
||
channel_chat_id: str,
|
||
conversation_event_id: UUID | None,
|
||
is_shadow: bool,
|
||
wait_seconds: int,
|
||
) -> None:
|
||
"""等待後查 run state，仍 waiting_tool 才發 interim"""
|
||
await asyncio.sleep(wait_seconds)
|
||
|
||
try:
|
||
from src.db.base import get_db_context
|
||
|
||
async with get_db_context(project_id) as db:
|
||
result = await db.execute(
|
||
select(AwoooPRunState.state, AwoooPRunState.is_shadow).where(
|
||
AwoooPRunState.run_id == run_id,
|
||
AwoooPRunState.project_id == project_id,
|
||
)
|
||
)
|
||
row = result.first()
|
||
|
||
if row is None:
|
||
logger.warning(
|
||
"interim_feedback_run_not_found",
|
||
run_id=str(run_id),
|
||
)
|
||
return
|
||
|
||
state, run_is_shadow = row
|
||
if state != "waiting_tool":
|
||
# run 已推進（complete/failed 等），不需要 interim
|
||
return
|
||
|
||
waiting_since = datetime.now(UTC)
|
||
interim_content = "AI 正在分析中，請稍候... ⏳"
|
||
|
||
await record_outbound_message(
|
||
db,
|
||
project_id=project_id,
|
||
run_id=run_id,
|
||
channel_type=channel_type,
|
||
channel_chat_id=channel_chat_id,
|
||
message_type="interim",
|
||
content=interim_content,
|
||
send_status="pending",
|
||
conversation_event_id=conversation_event_id,
|
||
triggered_by_state="waiting_tool",
|
||
waiting_since=waiting_since,
|
||
is_shadow=is_shadow or run_is_shadow,
|
||
)
|
||
|
||
if not (is_shadow or run_is_shadow):
|
||
# Non-shadow：實際發 Telegram 訊息
|
||
await _send_telegram_interim(
|
||
channel_chat_id=channel_chat_id,
|
||
content=interim_content,
|
||
run_id=run_id,
|
||
)
|
||
|
||
logger.info(
|
||
"interim_feedback_sent",
|
||
project_id=project_id,
|
||
run_id=str(run_id),
|
||
is_shadow=is_shadow or run_is_shadow,
|
||
)
|
||
|
||
except Exception as exc:
|
||
logger.exception(
|
||
"interim_feedback_task_error",
|
||
run_id=str(run_id),
|
||
error=str(exc),
|
||
)
|
||
|
||
|
||
async def _send_telegram_interim(
|
||
*,
|
||
channel_chat_id: str,
|
||
content: str,
|
||
run_id: UUID,
|
||
) -> None:
|
||
"""實際發送 Telegram interim 訊息（non-shadow 專用）"""
|
||
try:
|
||
import os
|
||
|
||
import httpx
|
||
|
||
bot_token = os.environ.get("TELEGRAM_BOT_TOKEN")
|
||
if not bot_token:
|
||
logger.warning("interim_telegram_no_token", run_id=str(run_id))
|
||
return
|
||
|
||
async with httpx.AsyncClient(timeout=10) as client:
|
||
await client.post(
|
||
f"https://api.telegram.org/bot{bot_token}/sendMessage",
|
||
json={
|
||
"chat_id": channel_chat_id,
|
||
"text": content,
|
||
"parse_mode": "HTML",
|
||
},
|
||
)
|
||
except Exception as exc:
|
||
logger.warning(
|
||
"interim_telegram_send_failed",
|
||
run_id=str(run_id),
|
||
error=str(exc),
|
||
)
|
||
|
||
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
# Channel Hub 主入口（Telegram inbound）
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
|
||
async def handle_telegram_inbound(
|
||
db: AsyncSession,
|
||
*,
|
||
project_id: str,
|
||
agent_id: str,
|
||
message_id: str,
|
||
user_id: str,
|
||
chat_id: str,
|
||
text: str | None = None,
|
||
is_shadow: bool = True,
|
||
) -> dict[str, Any]:
|
||
"""
|
||
Telegram 入站訊息的統一處理入口：
|
||
1. mirror_inbound_event（記錄）
|
||
2. create_run（建立 platform run）
|
||
3. schedule_interim_feedback（Progressive Feedback）
|
||
4. 回傳 {event_id, run_id, is_duplicate}
|
||
"""
|
||
# Step 1: 嘗試建立 run（有冪等保護）
|
||
run_id, is_duplicate = await create_run(
|
||
project_id=project_id,
|
||
agent_id=agent_id,
|
||
trigger_type="channel_event",
|
||
trigger_ref=f"telegram:{message_id}",
|
||
input_payload={"chat_id": chat_id, "user_id": user_id},
|
||
channel_type="telegram",
|
||
provider_event_id=message_id,
|
||
)
|
||
|
||
# Step 2: Mirror event（含 run_id）
|
||
event_id = await mirror_inbound_event(
|
||
db,
|
||
project_id=project_id,
|
||
channel_type="telegram",
|
||
provider_event_id=message_id,
|
||
channel_user_id=user_id,
|
||
channel_chat_id=chat_id,
|
||
content_type="text" if text else "callback_query",
|
||
raw_content=text,
|
||
run_id=run_id,
|
||
is_duplicate=is_duplicate,
|
||
)
|
||
|
||
# Step 3: Progressive Feedback（30s 計時器）
|
||
if not is_duplicate:
|
||
await schedule_interim_feedback(
|
||
project_id=project_id,
|
||
run_id=run_id,
|
||
channel_type="telegram",
|
||
channel_chat_id=chat_id,
|
||
conversation_event_id=event_id,
|
||
is_shadow=is_shadow,
|
||
)
|
||
|
||
return {
|
||
"event_id": str(event_id),
|
||
"run_id": str(run_id),
|
||
"is_duplicate": is_duplicate,
|
||
}
|