实施指南

技术研究人工智能 OpenCode

TELEGRAM_BOT_TOKEN=your-bot-token-from-botfather TELEGRAM_CHAT_ID=your-chat-id REPO_B_PATH=/path/to/repository-b OPENCODE_TIMEOUT=3600 超时时间（秒）

配置步骤

仓库A（Telegram Bot）配置

1. 环境变量配置

创建 .env 文件存储敏感信息：

# .env 文件
TELEGRAM_BOT_TOKEN="your-bot-token-from-botfather"
TELEGRAM_CHAT_ID="your-chat-id"
REPO_B_PATH="/path/to/repository-b"
OPENCODE_TIMEOUT=3600  # 超时时间（秒）

获取 Chat ID 的方法：

# 1. 向你的 Bot 发送任意消息
# 2. 使用以下命令获取 Chat ID
curl -s "https://api.telegram.org/bot${TELEGRAM_BOT_TOKEN}/getUpdates" | \
  jq '.result[0].message.chat.id'

2. Telegram Bot 消息处理器

创建 bot_handler.py：

#!/usr/bin/env python3
"""
Telegram Bot 消息处理器
处理 /research 命令并触发跨仓库研究任务
"""

import os
import subprocess
import sys
from telegram import Update
from telegram.ext import Application, CommandHandler, ContextTypes

# 加载环境变量
TELEGRAM_BOT_TOKEN = os.getenv('TELEGRAM_BOT_TOKEN')
REPO_B_PATH = os.getenv('REPO_B_PATH', '/path/to/repo-b')

async def research_command(update: Update, context: ContextTypes.DEFAULT_TYPE):
    """处理 /research 命令"""
    
    # 提取研究主题（命令后的所有文本）
    research_topic = ' '.join(context.args)
    
    if not research_topic:
        await update.message.reply_text(
            "请提供研究主题\n"
            "用法: /research <研究主题>\n"
            "示例: /research 帮我调研电动车行业的现状"
        )
        return
    
    # 发送确认消息
    await update.message.reply_text(
        f"开始研究: {research_topic}\n"
        f"请稍候，研究完成后将通知您..."
    )
    
    # 调用仓库B的包装脚本
    try:
        result = subprocess.run(
            ['bash', f'{REPO_B_PATH}/research_wrapper.sh', research_topic],
            capture_output=True,
            text=True,
            timeout=3600  # 1小时超时
        )
        
        if result.returncode == 0:
            await update.message.reply_text(
                f"研究完成: {research_topic}\n"
                f"报告已提交到 GitHub"
            )
        else:
            await update.message.reply_text(
                f"研究失败: {research_topic}\n"
                f"错误码: {result.returncode}\n"
                f"日志: {result.stderr[:500]}"
            )
    
    except subprocess.TimeoutExpired:
        await update.message.reply_text(
            f"研究超时: {research_topic}\n"
            f"任务执行超过1小时，已被终止"
        )
    
    except Exception as e:
        await update.message.reply_text(
            f"系统错误: {str(e)}"
        )

def main():
    """启动 Bot"""
    if not TELEGRAM_BOT_TOKEN:
        print("错误: 未设置 TELEGRAM_BOT_TOKEN 环境变量")
        sys.exit(1)
    
    application = Application.builder().token(TELEGRAM_BOT_TOKEN).build()
    
    # 注册命令处理器
    application.add_handler(CommandHandler("research", research_command))
    
    # 启动轮询
    print("Bot 已启动，等待消息...")
    application.run_polling()

if __name__ == '__main__':
    main()

安装依赖：

pip install python-telegram-bot

仓库B（Research）配置

1. 创建进程包装脚本

创建 research_wrapper.sh：

#!/bin/bash
#
# OpenCode 研究任务包装脚本
# 功能：执行研究、捕获状态、发送通知、清理进程
#

set -euo pipefail  # 严格模式

# ============================================================================
# 配置变量
# ============================================================================
TELEGRAM_BOT_TOKEN="${TELEGRAM_BOT_TOKEN:-}"
TELEGRAM_CHAT_ID="${TELEGRAM_CHAT_ID:-}"
LOG_FILE="${LOG_FILE:-/var/log/opencode_research.log}"
RESEARCH_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
LOCK_FILE="/tmp/opencode_research.lock"

# ============================================================================
# 日志函数
# ============================================================================
log() {
    local level="$1"
    shift
    local message="$*"
    local timestamp
    timestamp=$(date '+%Y-%m-%d %H:%M:%S')
    echo "[$timestamp] [$level] $message" | tee -a "$LOG_FILE"
}

# ============================================================================
# Telegram 通知函数
# ============================================================================
send_telegram_notification() {
    local status="$1"
    local message="$2"
    
    if [[ -z "$TELEGRAM_BOT_TOKEN" || -z "$TELEGRAM_CHAT_ID" ]]; then
        log "WARN" "Telegram 凭证未配置，跳过通知"
        return 0
    fi
    
    local emoji
    case "$status" in
        success) emoji="✅" ;;
        failure) emoji="❌" ;;
        timeout) emoji="⏰" ;;
        *) emoji="ℹ️" ;;
    esac
    
    local full_message="${emoji} ${message}"
    local api_url="https://api.telegram.org/bot${TELEGRAM_BOT_TOKEN}/sendMessage"
    
    # 发送通知（静默模式，不显示curl输出）
    curl -s -X POST "$api_url" \
        -d "chat_id=$TELEGRAM_CHAT_ID" \
        -d "text=$full_message" \
        -d "parse_mode=HTML" \
        > /dev/null 2>&1 || {
        log "ERROR" "发送 Telegram 通知失败"
    }
}

# ============================================================================
# 进程管理函数
# ============================================================================
find_opencode_processes() {
    # 查找当前用户的 opencode 进程（排除当前shell和grep本身）
    pgrep -u "$(whoami)" -f "opencode.*run" || true
}

cleanup_opencode() {
    log "INFO" "开始清理 opencode 进程..."
    
    local pids
    pids=$(find_opencode_processes)
    
    if [[ -n "$pids" ]]; then
        log "INFO" "发现 opencode 进程: $pids"
        
        # 先尝试优雅终止（SIGTERM）
        echo "$pids" | xargs -r kill -TERM 2>/dev/null || true
        
        # 等待3秒
        sleep 3
        
        # 检查是否仍然存在，如果存在则强制终止（SIGKILL）
        local remaining_pids
        remaining_pids=$(find_opencode_processes)
        
        if [[ -n "$remaining_pids" ]]; then
            log "WARN" "进程未响应 SIGTERM，使用 SIGKILL"
            echo "$remaining_pids" | xargs -r kill -KILL 2>/dev/null || true
        fi
        
        log "INFO" "进程清理完成"
    else
        log "INFO" "未发现需要清理的 opencode 进程"
    fi
}

start_new_session() {
    log "INFO" "启动新的 OpenCode 会话..."
    
    # 方式1: 启动新的 opencode TUI（后台运行）
    # opencode &
    
    # 方式2: 如果只是想确保环境干净，可以仅清理进程
    cleanup_opencode
    
    log "INFO" "新会话准备就绪"
}

# ============================================================================
# 主函数
# ============================================================================
main() {
    local research_topic="$1"
    local start_time
    start_time=$(date +%s)
    
    log "INFO" "=========================================="
    log "INFO" "开始研究任务: $research_topic"
    log "INFO" "工作目录: $RESEARCH_DIR"
    log "INFO" "=========================================="
    
    # 检查锁定（防止并发执行）
    if [[ -f "$LOCK_FILE" ]]; then
        local lock_pid
        lock_pid=$(cat "$LOCK_FILE" 2>/dev/null || echo "unknown")
        log "ERROR" "另一个研究任务正在运行 (PID: $lock_pid)"
        send_telegram_notification "failure" "研究任务冲突: 另一个任务正在运行"
        exit 1
    fi
    
    # 创建锁定文件
    echo "$$" > "$LOCK_FILE"
    trap 'rm -f "$LOCK_FILE"' EXIT
    
    # 切换到仓库目录
    cd "$RESEARCH_DIR"
    
    # 执行研究任务
    log "INFO" "执行: opencode run \"$research_topic\""
    
    local exit_code=0
    local opencode_output
    
    # 捕获 opencode 输出和退出码
    if opencode_output=$(opencode run "$research_topic" 2>&1); then
        exit_code=0
        log "INFO" "研究任务成功完成"
    else
        exit_code=$?
        log "ERROR" "研究任务失败，退出码: $exit_code"
    fi
    
    # 记录输出（限制长度）
    echo "$opencode_output" | tail -n 100 >> "$LOG_FILE"
    
    # 计算执行时间
    local end_time
    end_time=$(date +%s)
    local duration=$((end_time - start_time))
    local duration_str="$((duration / 60))分$((duration % 60))秒"
    
    # 发送通知
    if [[ $exit_code -eq 0 ]]; then
        send_telegram_notification "success" \
            "研究完成: $research_topic\n" \
            "耗时: $duration_str\n" \
            "报告已提交到 GitHub"
    else
        send_telegram_notification "failure" \
            "研究失败: $research_topic\n" \
            "耗时: $duration_str\n" \
            "退出码: $exit_code"
    fi
    
    # 可选：清理进程或启动新会话
    # cleanup_opencode
    # start_new_session
    
    log "INFO" "=========================================="
    log "INFO" "研究任务结束，总耗时: $duration_str"
    log "INFO" "=========================================="
    
    exit $exit_code
}

# ============================================================================
# 脚本入口
# ============================================================================

# 检查参数
if [[ $# -lt 1 ]]; then
    echo "用法: $0 <研究主题>"
    echo "示例: $0 \"帮我调研电动车行业的现状\""
    exit 1
fi

# 确保日志目录存在
mkdir -p "$(dirname "$LOG_FILE")"

# 运行主函数
main "$@"

设置执行权限：

chmod +x research_wrapper.sh

2. GitHub 自动提交配置

创建 github_autocommit.sh：

#!/bin/bash
#
# 研究完成后自动提交到 GitHub
#

set -e

RESEARCH_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
cd "$RESEARCH_DIR"

# 配置 Git（如果尚未配置）
git config user.email "research-bot@example.com" || true
git config user.name "Research Bot" || true

# 检查是否有变更
if [[ -n $(git status --porcelain) ]]; then
    echo "检测到文件变更，开始提交..."
    
    # 添加所有变更
    git add -A
    
    # 创建提交
    local timestamp
    timestamp=$(date '+%Y-%m-%d %H:%M:%S')
    git commit -m "research: auto-update research results [${timestamp}]"
    
    # 推送到 GitHub
    git push origin main || git push origin master
    
    echo "✅ 已成功提交到 GitHub"
else
    echo "无文件变更，跳过提交"
fi

注意：确保仓库B已配置 GitHub 凭证（SSH key 或 Personal Access Token）。

代码片段

方案B1：增强版 Shell 包装器

包含更多错误处理和日志记录：

#!/bin/bash
#
# 增强版 OpenCode 研究包装器
# 特性：超时控制、详细日志、自动重试
#

set -euo pipefail

# 配置
RESEARCH_TOPIC="${1:-}"
TIMEOUT_SECONDS="${TIMEOUT_SECONDS:-1800}"  # 默认30分钟超时
MAX_RETRIES="${MAX_RETRIES:-2}"
RETRY_COUNT=0
TELEGRAM_BOT_TOKEN="${TELEGRAM_BOT_TOKEN:-}"
TELEGRAM_CHAT_ID="${TELEGRAM_CHAT_ID:-}"

# 颜色输出
RED='\033[0;31m'
GREEN='\033[0;32m'
YELLOW='\033[1;33m'
NC='\033[0m' # No Color

log_info() {
    echo -e "${GREEN}[INFO]${NC} $*"
}

log_warn() {
    echo -e "${YELLOW}[WARN]${NC} $*"
}

log_error() {
    echo -e "${RED}[ERROR]${NC} $*"
}

send_notification() {
    local status="$1"
    local message="$2"
    
    [[ -z "$TELEGRAM_BOT_TOKEN" ]] && return 0
    
    curl -s -X POST "https://api.telegram.org/bot${TELEGRAM_BOT_TOKEN}/sendMessage" \
        -d "chat_id=$TELEGRAM_CHAT_ID" \
        -d "text=$message" \
        > /dev/null
}

execute_research() {
    local topic="$1"
    local temp_log
    temp_log=$(mktemp)
    
    log_info "开始研究: $topic"
    log_info "超时设置: ${TIMEOUT_SECONDS}秒"
    
    # 使用 timeout 命令控制执行时间
    if timeout "$TIMEOUT_SECONDS" opencode run "$topic" > "$temp_log" 2>&1; then
        log_info "研究成功完成"
        send_notification "success" "✅ 研究完成: $topic"
        rm -f "$temp_log"
        return 0
    else
        local exit_code=$?
        
        if [[ $exit_code -eq 124 ]]; then
            log_error "研究超时（${TIMEOUT_SECONDS}秒）"
            send_notification "timeout" "⏰ 研究超时: $topic"
        else
            log_error "研究失败，退出码: $exit_code"
            log_error "错误输出:"
            cat "$temp_log"
            send_notification "failure" "❌ 研究失败: $topic (退出码: $exit_code)"
        fi
        
        rm -f "$temp_log"
        return $exit_code
    fi
}

# 主逻辑
if [[ -z "$RESEARCH_TOPIC" ]]; then
    log_error "请提供研究主题"
    exit 1
fi

# 重试循环
while [[ $RETRY_COUNT -lt $MAX_RETRIES ]]; do
    if execute_research "$RESEARCH_TOPIC"; then
        exit 0
    fi
    
    RETRY_COUNT=$((RETRY_COUNT + 1))
    
    if [[ $RETRY_COUNT -lt $MAX_RETRIES ]]; then
        log_warn "重试 $RETRY_COUNT/$MAX_RETRIES..."
        sleep 5
    fi
done

log_error "达到最大重试次数，任务失败"
exit 1

方案B2：文件信号机制（Node.js 实现）

如果仓库A使用 Node.js：

// fileWatcher.js
const fs = require('fs');
const path = require('path');
const { exec } = require('child_process');

const SIGNAL_FILE = '/tmp/research_complete.signal';
const TELEGRAM_BOT_TOKEN = process.env.TELEGRAM_BOT_TOKEN;
const TELEGRAM_CHAT_ID = process.env.TELEGRAM_CHAT_ID;

// 发送 Telegram 通知
async function sendTelegramNotification(message) {
    if (!TELEGRAM_BOT_TOKEN || !TELEGRAM_CHAT_ID) {
        console.warn('Telegram 凭证未配置');
        return;
    }
    
    const url = `https://api.telegram.org/bot${TELEGRAM_BOT_TOKEN}/sendMessage`;
    const params = new URLSearchParams({
        chat_id: TELEGRAM_CHAT_ID,
        text: message,
        parse_mode: 'HTML'
    });
    
    try {
        await fetch(`${url}?${params}`);
    } catch (error) {
        console.error('发送通知失败:', error);
    }
}

// 处理研究完成信号
async function handleResearchComplete(status) {
    console.log(`研究完成，状态: ${status}`);
    
    if (status === 'success') {
        await sendTelegramNotification('✅ 研究任务成功完成');
        
        // 执行清理
        exec('pkill -f "opencode.*run"', (error) => {
            if (error) console.error('清理进程失败:', error);
            else console.log('进程已清理');
        });
    } else {
        await sendTelegramNotification(`❌ 研究任务失败: ${status}`);
    }
    
    // 清理信号文件
    try {
        fs.unlinkSync(SIGNAL_FILE);
    } catch (error) {
        // 忽略错误
    }
}

// 监控文件变化
function startWatching() {
    console.log(`开始监控信号文件: ${SIGNAL_FILE}`);
    
    // 使用 fs.watchFile 轮询方式（兼容性更好）
    fs.watchFile(SIGNAL_FILE, { interval: 1000 }, (curr, prev) => {
        if (curr.size > 0 && curr.mtime !== prev.mtime) {
            try {
                const status = fs.readFileSync(SIGNAL_FILE, 'utf8').trim();
                handleResearchComplete(status);
            } catch (error) {
                console.error('读取信号文件失败:', error);
            }
        }
    });
    
    console.log('文件监控已启动');
}

// 启动监控
startWatching();

// 保持进程运行
process.stdin.resume();

方案B3：Redis 消息队列（Python 实现）

#!/usr/bin/env python3
"""
Redis 协调器 - 研究任务完成通知
"""

import json
import redis
import subprocess
import os

# 配置
REDIS_HOST = os.getenv('REDIS_HOST', 'localhost')
REDIS_PORT = int(os.getenv('REDIS_PORT', 6379))
REDIS_CHANNEL = 'opencode_research'
TELEGRAM_BOT_TOKEN = os.getenv('TELEGRAM_BOT_TOKEN')
TELEGRAM_CHAT_ID = os.getenv('TELEGRAM_CHAT_ID')

def send_telegram_notification(message: str):
    """发送 Telegram 通知"""
    if not TELEGRAM_BOT_TOKEN or not TELEGRAM_CHAT_ID:
        print("警告: Telegram 凭证未配置")
        return
    
    import urllib.request
    import urllib.parse
    
    url = f"https://api.telegram.org/bot{TELEGRAM_BOT_TOKEN}/sendMessage"
    data = urllib.parse.urlencode({
        'chat_id': TELEGRAM_CHAT_ID,
        'text': message,
        'parse_mode': 'HTML'
    }).encode()
    
    try:
        urllib.request.urlopen(url, data=data, timeout=10)
    except Exception as e:
        print(f"发送通知失败: {e}")

def handle_research_complete(data: dict):
    """处理研究完成消息"""
    status = data.get('status', 'unknown')
    topic = data.get('topic', '未知主题')
    duration = data.get('duration', 0)
    
    if status == 'success':
        send_telegram_notification(
            f"✅ <b>研究完成</b>\n"
            f"主题: {topic}\n"
            f"耗时: {duration}秒"
        )
        
        # 执行清理
        subprocess.run(['pkill', '-f', 'opencode.*run'], capture_output=True)
        
    else:
        error = data.get('error', '未知错误')
        send_telegram_notification(
            f"❌ <b>研究失败</b>\n"
            f"主题: {topic}\n"
            f"错误: {error}"
        )

def main():
    """主函数"""
    print(f"连接到 Redis: {REDIS_HOST}:{REDIS_PORT}")
    
    r = redis.Redis(host=REDIS_HOST, port=REDIS_PORT, decode_responses=True)
    
    # 测试连接
    try:
        r.ping()
        print("Redis 连接成功")
    except redis.ConnectionError:
        print("Redis 连接失败，请检查配置")
        return
    
    # 订阅频道
    pubsub = r.pubsub()
    pubsub.subscribe(REDIS_CHANNEL)
    
    print(f"已订阅频道: {REDIS_CHANNEL}")
    print("等待消息...")
    
    for message in pubsub.listen():
        if message['type'] == 'message':
            try:
                data = json.loads(message['data'])
                print(f"收到消息: {data}")
                
                if data.get('type') == 'research_complete':
                    handle_research_complete(data)
                    
            except json.JSONDecodeError:
                print(f"无效的 JSON 消息: {message['data']}")
            except Exception as e:
                print(f"处理消息时出错: {e}")

if __name__ == '__main__':
    main()

安装依赖：

pip install redis

部署清单

环境准备

Linux/Unix 服务器（Ubuntu 20.04+ 推荐）
Node.js 14+ 和 npm（如果仓库A使用 Node.js）
Python 3.8+ 和 pip（如果仓库A使用 Python）
OpenCode CLI 已安装并配置
Git 已配置（用于仓库B自动提交）

依赖安装

# 基础工具
sudo apt-get update
sudo apt-get install -y curl jq git

# Python 依赖（如果使用 Python Bot）
pip install python-telegram-bot redis

# Node.js 依赖（如果使用 Node.js Bot）
npm install node-telegram-bot-api

配置检查清单

Telegram Bot Token 已获取（从 @BotFather）
Telegram Chat ID 已获取
环境变量已配置（.env 文件或系统环境变量）
仓库B路径正确配置
OpenCode CLI 认证已配置（opencode auth login）
GitHub SSH key 或 Token 已配置（用于自动提交）
日志目录权限已设置（mkdir -p /var/log && chmod 755 /var/log）

测试步骤

# 1. 测试 Telegram Bot
curl -s "https://api.telegram.org/bot${TELEGRAM_BOT_TOKEN}/getMe"

# 2. 测试 OpenCode 执行
opencode run "echo hello"

# 3. 测试包装脚本
bash research_wrapper.sh "测试研究主题"

# 4. 测试 GitHub 提交
cd /path/to/repo-b
bash github_autocommit.sh

# 5. 端到端测试（通过 Telegram 发送 /research 命令）

监控与日志

查看日志：

# 实时查看研究日志
tail -f /var/log/opencode_research.log

# 查看 Bot 日志
journalctl -u telegram-bot -f

Systemd 服务配置（可选）：

创建 /etc/systemd/system/telegram-bot.service：

[Unit]
Description=Telegram Research Bot
After=network.target

[Service]
Type=simple
User=ubuntu
WorkingDirectory=/path/to/repo-a
Environment="TELEGRAM_BOT_TOKEN=your-token"
Environment="TELEGRAM_CHAT_ID=your-chat-id"
ExecStart=/usr/bin/python3 /path/to/repo-a/bot_handler.py
Restart=always
RestartSec=10

[Install]
WantedBy=multi-user.target

启动服务：

sudo systemctl daemon-reload
sudo systemctl enable telegram-bot
sudo systemctl start telegram-bot
sudo systemctl status telegram-bot

故障排除

常见问题

Q: opencode 进程没有被清理

检查 pkill 命令的匹配模式是否正确
使用 ps aux | grep opencode 查看进程
尝试使用更精确的匹配：pkill -f "opencode run"

Q: Telegram 通知没有收到

检查 TELEGRAM_BOT_TOKEN 和 TELEGRAM_CHAT_ID 是否正确
使用 curl 直接测试 API：curl "https://api.telegram.org/bot${TOKEN}/getMe"
检查 Bot 是否已启动并与用户对话（发送 /start）

Q: 研究任务超时

增加 TIMEOUT_SECONDS 环境变量
检查 opencode 是否卡住（查看日志中的输出）
考虑使用后台执行 + 异步通知模式

参考资料

Telegram Bot API Documentation - sendMessage endpoint
Python Telegram Bot Library - Bot framework
Bash Timeout Command - Process timeout control
Redis Pub/Sub - Message queue implementation
GitHub Actions - Alternative CI/CD approach