422 lines
17 KiB
Python
422 lines
17 KiB
Python
"""Discord bot — relays Antigravity brain events to Discord channels.
|
||
|
||
Dynamic channel management:
|
||
- Creates `AG-{project_name}` channels only when file events arrive
|
||
- NO startup channel creation — only reconnects to existing Discord channels
|
||
- Archives channels after 10 minutes of inactivity
|
||
"""
|
||
|
||
import asyncio
|
||
import json
|
||
import logging
|
||
import re
|
||
import time
|
||
from datetime import datetime, timezone
|
||
from pathlib import Path
|
||
|
||
import discord
|
||
from discord.ext import commands, tasks
|
||
|
||
from config import Config
|
||
from parser import (
|
||
parse_task_progress,
|
||
md_to_discord_text,
|
||
format_task_embed_text,
|
||
)
|
||
from watcher import BrainEvent, EventType
|
||
from bridge import BridgeProtocol, ApprovalRequest, UserResponse
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
|
||
# ─── Discord UI Components ──────────────────────────────────────────
|
||
|
||
class ApprovalView(discord.ui.View):
|
||
"""Discord buttons for approving/rejecting Antigravity actions."""
|
||
|
||
def __init__(self, bridge: BridgeProtocol, request: ApprovalRequest):
|
||
super().__init__(timeout=300)
|
||
self.bridge = bridge
|
||
self.request = request
|
||
self.responded = False
|
||
|
||
@discord.ui.button(label="✅ 승인", style=discord.ButtonStyle.green)
|
||
async def approve(self, interaction: discord.Interaction, button: discord.ui.Button):
|
||
if self.responded:
|
||
await interaction.response.send_message("이미 응답됨", ephemeral=True)
|
||
return
|
||
self.responded = True
|
||
self.bridge.write_response(UserResponse(
|
||
request_id=self.request.request_id, approved=True,
|
||
))
|
||
embed = interaction.message.embeds[0] if interaction.message.embeds else None
|
||
if embed:
|
||
embed.color = discord.Color.green()
|
||
embed.set_footer(text=f"✅ 승인됨 by {interaction.user.display_name}")
|
||
await interaction.response.edit_message(embed=embed, view=None)
|
||
|
||
@discord.ui.button(label="❌ 거부", style=discord.ButtonStyle.red)
|
||
async def reject(self, interaction: discord.Interaction, button: discord.ui.Button):
|
||
if self.responded:
|
||
await interaction.response.send_message("이미 응답됨", ephemeral=True)
|
||
return
|
||
self.responded = True
|
||
self.bridge.write_response(UserResponse(
|
||
request_id=self.request.request_id, approved=False,
|
||
))
|
||
embed = interaction.message.embeds[0] if interaction.message.embeds else None
|
||
if embed:
|
||
embed.color = discord.Color.red()
|
||
embed.set_footer(text=f"❌ 거부됨 by {interaction.user.display_name}")
|
||
await interaction.response.edit_message(embed=embed, view=None)
|
||
|
||
async def on_timeout(self):
|
||
if not self.responded:
|
||
self.bridge.write_response(UserResponse(
|
||
request_id=self.request.request_id, approved=False,
|
||
))
|
||
|
||
|
||
# ─── Project Name Detection ─────────────────────────────────────────
|
||
|
||
def detect_project_name(conv_dir: Path) -> str:
|
||
"""Extract project name from conversation artifacts.
|
||
Output: lowercase_with_underscores (e.g. 'gravity_control')
|
||
"""
|
||
short_id = conv_dir.name[:8]
|
||
|
||
def _sanitize(raw: str) -> str:
|
||
for suffix in ["Task Tracker", "— Task Tracker", "태스크", "구현 계획"]:
|
||
raw = raw.replace(suffix, "")
|
||
raw = raw.strip(" —-–")
|
||
raw = re.sub(r'[^a-zA-Z0-9가-힣\s\-]', '', raw)
|
||
raw = re.sub(r'[\s\-]+', '_', raw).strip('_').lower()
|
||
return raw[:30] if raw else ""
|
||
|
||
for fname in ["task.md", "implementation_plan.md"]:
|
||
fpath = conv_dir / fname
|
||
if fpath.exists():
|
||
try:
|
||
first_lines = fpath.read_text(encoding="utf-8").splitlines()[:5]
|
||
for line in first_lines:
|
||
match = re.match(r'^#\s+(.+)', line)
|
||
if match:
|
||
name = _sanitize(match.group(1))
|
||
if name and name != "task":
|
||
return name
|
||
except (OSError, UnicodeDecodeError):
|
||
pass
|
||
|
||
return short_id
|
||
|
||
|
||
# ─── Bot ─────────────────────────────────────────────────────────────
|
||
|
||
class GravityBot(commands.Bot):
|
||
"""Discord bot for Antigravity session monitoring."""
|
||
|
||
def __init__(self, event_queue: asyncio.Queue):
|
||
intents = discord.Intents.default()
|
||
intents.message_content = True
|
||
intents.guilds = True
|
||
super().__init__(command_prefix="!", intents=intents)
|
||
|
||
self.event_queue = event_queue
|
||
self.session_channels: dict[str, discord.TextChannel] = {}
|
||
self.session_status_messages: dict[str, int] = {}
|
||
self.session_names: dict[str, str] = {}
|
||
self._channel_create_lock = asyncio.Lock() # SINGLE global lock
|
||
self.bridge = BridgeProtocol()
|
||
self.session_category: discord.CategoryChannel | None = None
|
||
self.guild: discord.Guild | None = None
|
||
|
||
async def setup_hook(self):
|
||
self.loop.create_task(self._process_events())
|
||
self.pending_approval_scanner.start()
|
||
logger.info("Bot setup complete, event processor started")
|
||
|
||
async def on_ready(self):
|
||
logger.info(f"Bot connected as {self.user} (ID: {self.user.id})")
|
||
|
||
self.guild = self.get_guild(Config.DISCORD_GUILD_ID)
|
||
if not self.guild:
|
||
logger.error(f"Guild {Config.DISCORD_GUILD_ID} not found!")
|
||
return
|
||
|
||
# Find or create category
|
||
category_name = "Antigravity Sessions"
|
||
self.session_category = discord.utils.get(
|
||
self.guild.categories, name=category_name
|
||
)
|
||
if not self.session_category:
|
||
try:
|
||
self.session_category = await self.guild.create_category(category_name)
|
||
logger.info(f"Created category: {category_name}")
|
||
except discord.errors.Forbidden:
|
||
logger.error("No permission to create category!")
|
||
return
|
||
|
||
# ONLY reconnect to existing Discord channels (NO new creation)
|
||
await self._reconnect_existing_channels()
|
||
|
||
async def _reconnect_existing_channels(self):
|
||
"""Scan existing Discord channels and map them to conversation IDs."""
|
||
if not self.session_category:
|
||
return
|
||
|
||
count = 0
|
||
for ch in self.session_category.text_channels:
|
||
if ch.topic and "Antigravity Session:" in ch.topic:
|
||
conv_id = ch.topic.replace("Antigravity Session:", "").strip()
|
||
if conv_id:
|
||
self.session_channels[conv_id] = ch
|
||
await self._recover_task_message(ch, conv_id)
|
||
count += 1
|
||
|
||
logger.info(f"Reconnected to {count} existing channels")
|
||
|
||
async def _recover_task_message(
|
||
self, channel: discord.TextChannel, conversation_id: str
|
||
):
|
||
if conversation_id in self.session_status_messages:
|
||
return
|
||
try:
|
||
async for msg in channel.history(limit=10):
|
||
if msg.author == self.user and msg.embeds:
|
||
embed = msg.embeds[0]
|
||
if embed.title and "Task" in embed.title:
|
||
self.session_status_messages[conversation_id] = msg.id
|
||
return
|
||
except (discord.Forbidden, discord.HTTPException):
|
||
pass
|
||
|
||
# ─── Channel Management ──────────────────────────────────────────
|
||
|
||
async def _ensure_channel(
|
||
self, conversation_id: str, project_name: str
|
||
) -> discord.TextChannel:
|
||
"""Get or create a channel. Uses GLOBAL lock + OWN dict lookup (no cache issues)."""
|
||
# Fast path: already mapped
|
||
if conversation_id in self.session_channels:
|
||
return self.session_channels[conversation_id]
|
||
|
||
async with self._channel_create_lock:
|
||
# Double-check after lock
|
||
if conversation_id in self.session_channels:
|
||
return self.session_channels[conversation_id]
|
||
|
||
channel_name = f"{Config.CHANNEL_PREFIX}-{project_name}"
|
||
target_name = channel_name.lower().replace(" ", "-")
|
||
|
||
# Check OWN dict for a channel with the same NAME
|
||
# (different conv IDs can share one channel)
|
||
for cid, ch in self.session_channels.items():
|
||
if ch.name == target_name:
|
||
self.session_channels[conversation_id] = ch
|
||
self.session_names[conversation_id] = project_name
|
||
logger.info(f"Reusing channel #{ch.name} for {conversation_id[:8]}")
|
||
return ch
|
||
|
||
# Create new channel (truly first time)
|
||
try:
|
||
channel = await self.guild.create_text_channel(
|
||
name=channel_name,
|
||
category=self.session_category,
|
||
topic=f"Antigravity Session: {conversation_id}",
|
||
)
|
||
self.session_channels[conversation_id] = channel
|
||
self.session_names[conversation_id] = project_name
|
||
logger.info(f"Created channel #{channel_name}")
|
||
|
||
embed = discord.Embed(
|
||
title=f"🚀 {project_name}",
|
||
description=f"Antigravity 세션 연결됨\nSession: `{conversation_id}`",
|
||
color=discord.Color.blue(),
|
||
timestamp=datetime.now(timezone.utc),
|
||
)
|
||
await channel.send(embed=embed)
|
||
return channel
|
||
|
||
except discord.errors.Forbidden:
|
||
logger.error(f"No permission to create channel: {channel_name}")
|
||
return None
|
||
|
||
# ─── Event Processing (SINGLE ROUTE) ─────────────────────────────
|
||
|
||
async def _process_events(self):
|
||
"""Main event loop — ALL events go through here sequentially."""
|
||
await self.wait_until_ready()
|
||
|
||
while not self.is_closed():
|
||
try:
|
||
event = await asyncio.wait_for(
|
||
self.event_queue.get(), timeout=5.0
|
||
)
|
||
await self._handle_event(event)
|
||
except asyncio.TimeoutError:
|
||
continue
|
||
except Exception as e:
|
||
logger.error(f"Error processing event: {e}", exc_info=True)
|
||
|
||
async def _handle_event(self, event: BrainEvent):
|
||
"""Route brain events to handlers — single entry point."""
|
||
conv_dir = Config.BRAIN_PATH / event.conversation_id
|
||
project_name = detect_project_name(conv_dir)
|
||
|
||
if event.event_type == EventType.SESSION_START:
|
||
await self._ensure_channel(event.conversation_id, project_name)
|
||
return
|
||
|
||
# FILE_CREATED or FILE_CHANGED
|
||
channel = await self._ensure_channel(event.conversation_id, project_name)
|
||
if not channel:
|
||
return
|
||
|
||
try:
|
||
if event.file_name == "task.md":
|
||
await self._send_task_update(channel, event)
|
||
else:
|
||
await self._send_artifact_update(channel, event)
|
||
except discord.NotFound:
|
||
# Channel was deleted while we held a reference
|
||
self.session_channels.pop(event.conversation_id, None)
|
||
self.session_status_messages.pop(event.conversation_id, None)
|
||
logger.warning(f"Channel deleted, cleared: {event.conversation_id[:8]}")
|
||
|
||
# ─── Message Senders ─────────────────────────────────────────────
|
||
|
||
async def _send_task_update(
|
||
self, channel: discord.TextChannel, event: BrainEvent
|
||
):
|
||
"""Send/edit task progress embed (SINGLE message, always edited)."""
|
||
progress = parse_task_progress(event.content)
|
||
|
||
embed = discord.Embed(
|
||
title="📋 Task 진행 현황",
|
||
description=format_task_embed_text(progress),
|
||
color=discord.Color.gold() if progress.in_progress > 0
|
||
else discord.Color.green() if progress.done == progress.total
|
||
else discord.Color.greyple(),
|
||
timestamp=datetime.now(timezone.utc),
|
||
)
|
||
embed.set_footer(text=f"Session: {event.conversation_id[:8]}")
|
||
|
||
# Always try to edit existing message first
|
||
msg_id = self.session_status_messages.get(event.conversation_id)
|
||
if msg_id:
|
||
try:
|
||
msg = await channel.fetch_message(msg_id)
|
||
await msg.edit(embed=embed)
|
||
return
|
||
except (discord.NotFound, discord.HTTPException):
|
||
pass
|
||
|
||
msg = await channel.send(embed=embed)
|
||
self.session_status_messages[event.conversation_id] = msg.id
|
||
|
||
async def _send_artifact_update(
|
||
self, channel: discord.TextChannel, event: BrainEvent
|
||
):
|
||
"""Send artifact update as single compact embed (preview only)."""
|
||
labels = {
|
||
"implementation_plan.md": "📐 구현 계획",
|
||
"walkthrough.md": "📝 작업 결과 요약",
|
||
}
|
||
label = labels.get(event.file_name, f"📄 {event.file_name}")
|
||
event_label = "생성" if event.event_type == EventType.FILE_CREATED else "업데이트"
|
||
|
||
# Preview: first 6 non-empty lines only
|
||
lines = event.content.strip().splitlines()
|
||
preview = "\n".join(l for l in lines[:6] if l.strip())
|
||
if len(lines) > 6:
|
||
preview += f"\n... (+{len(lines) - 6} lines)"
|
||
|
||
embed = discord.Embed(
|
||
title=f"{label} ({event_label}됨)",
|
||
description=preview[:1000],
|
||
color=discord.Color.blue(),
|
||
timestamp=datetime.now(timezone.utc),
|
||
)
|
||
await channel.send(embed=embed)
|
||
|
||
# ─── Approval Scanner ────────────────────────────────────────────
|
||
|
||
@tasks.loop(seconds=3)
|
||
async def pending_approval_scanner(self):
|
||
"""Scan bridge/pending/ for new approval requests."""
|
||
try:
|
||
requests = self.bridge.get_pending_requests()
|
||
for req in requests:
|
||
if req.discord_message_id != 0:
|
||
continue # Already sent
|
||
|
||
channel = self.session_channels.get(req.conversation_id)
|
||
if not channel:
|
||
conv_dir = Config.BRAIN_PATH / req.conversation_id
|
||
if conv_dir.exists():
|
||
project_name = detect_project_name(conv_dir)
|
||
channel = await self._ensure_channel(
|
||
req.conversation_id, project_name
|
||
)
|
||
|
||
if channel:
|
||
await self._send_approval_request(channel, req)
|
||
except Exception as e:
|
||
logger.error(f"Error scanning approvals: {e}")
|
||
|
||
@pending_approval_scanner.before_loop
|
||
async def before_scanner(self):
|
||
await self.wait_until_ready()
|
||
|
||
async def _send_approval_request(
|
||
self, channel: discord.TextChannel, request: ApprovalRequest
|
||
):
|
||
embed = discord.Embed(
|
||
title="⚠️ 승인 요청",
|
||
description=(
|
||
f"**명령어:**\n```\n{request.command[:1000]}\n```\n"
|
||
f"{request.description[:500]}"
|
||
),
|
||
color=discord.Color.orange(),
|
||
timestamp=datetime.now(timezone.utc),
|
||
)
|
||
embed.set_footer(text=f"ID: {request.request_id}")
|
||
|
||
view = ApprovalView(self.bridge, request)
|
||
msg = await channel.send(embed=embed, view=view)
|
||
|
||
# Update pending file with discord message id
|
||
pending_file = self.bridge.pending_dir / f"{request.request_id}.json"
|
||
if pending_file.exists():
|
||
try:
|
||
data = json.loads(pending_file.read_text(encoding="utf-8"))
|
||
data["discord_message_id"] = msg.id
|
||
pending_file.write_text(
|
||
json.dumps(data, ensure_ascii=False, indent=2), encoding="utf-8"
|
||
)
|
||
except (json.JSONDecodeError, OSError):
|
||
pass
|
||
|
||
logger.info(f"Sent approval request: {request.request_id[:8]}")
|
||
|
||
# ─── Discord → Antigravity Text Relay ─────────────────────────────
|
||
|
||
async def on_message(self, message: discord.Message):
|
||
if message.author == self.user:
|
||
return
|
||
|
||
if not message.channel.name.startswith(Config.CHANNEL_PREFIX.lower() + "-"):
|
||
return
|
||
|
||
conv_id = None
|
||
for cid, ch in self.session_channels.items():
|
||
if ch.id == message.channel.id:
|
||
conv_id = cid
|
||
break
|
||
|
||
if conv_id and message.content.strip():
|
||
self.bridge.write_command(conv_id, message.content.strip())
|
||
await message.add_reaction("📨")
|
||
|
||
await self.process_commands(message)
|