OpenAI Codex「持久模式」深度解析：永不休眠的AI智能体如何重塑自动化边界

"""
Codex持久模式 - 任务调度与状态管理示例
基于OpenAI Codex SDK的异步任务调度机制
"""

from enum import Enum
from dataclasses import dataclass, field
from typing import Optional, List, Dict, Any
from datetime import datetime
import asyncio
import json
import uuid


class TaskStatus(Enum):
    """任务状态枚举"""
    PENDING = "pending"
    RUNNING = "running"
    COMPLETED = "completed"
    FAILED = "failed"
    SLEEPING = "sleeping"


class TaskPriority(Enum):
    """任务优先级"""
    LOW = 0
    MEDIUM = 1
    HIGH = 2
    CRITICAL = 3


@dataclass
class FollowUpTask:
    """后续任务数据结构"""
    id: str = field(default_factory=lambda: uuid.uuid4().hex[:12])
    description: str = ""
    priority: TaskPriority = TaskPriority.MEDIUM
    status: TaskStatus = TaskStatus.PENDING
    parent_task_id: Optional[str] = None
    created_at: datetime = field(default_factory=datetime.now)
    context_summary: str = ""
    dependencies: List[str] = field(default_factory=list)
    max_retries: int = 3
    retry_count: int = 0
    result: Optional[Dict[str, Any]] = None


class PersistentScheduler:
    """
    持久模式任务调度器
    管理任务的创建、调度、执行和状态转换
    """
    
    def __init__(self, max_concurrent: int = 3):
        self.task_queue: asyncio.PriorityQueue = asyncio.PriorityQueue()
        self.active_tasks: Dict[str, asyncio.Task] = {}
        self.completed_tasks: List[FollowUpTask] = []
        self.max_concurrent = max_concurrent
        self.semaphore = asyncio.Semaphore(max_concurrent)
        self._running = False
        
    async def create_follow_up(self, 
                                description: str,
                                priority: TaskPriority = TaskPriority.MEDIUM,
                                context: str = "",
                                parent_id: Optional[str] = None) -> str:
        """
        创建后续任务 - 这是持久模式的核心机制
        Codex在完成用户请求后自动调用此方法
        """
        task = FollowUpTask(
            description=description,
            priority=priority,
            context_summary=context,
            parent_task_id=parent_id
        )
        # 优先级队列：数值越小优先级越高
        await self.task_queue.put((-priority.value, task))
        print(f"[{datetime.now()}] 创建后续任务: {task.id}")
        print(f"  描述: {description[:80]}...")
        print(f"  优先级: {priority.name}")
        return task.id
    
    async def execute_task(self, task: FollowUpTask) -> Dict[str, Any]:
        """执行单个任务"""
        async with self.semaphore:
            task.status = TaskStatus.RUNNING
            print(f"[{datetime.now()}] 开始执行任务: {task.id}")
            
            try:
                # 模拟o3-mini模型的推理过程
                result = await self._invoke_model(task)
                task.status = TaskStatus.COMPLETED
                task.result = result
                print(f"[{datetime.now()}] 任务完成: {task.id}")
                return result
                
            except Exception as e:
                task.retry_count += 1
                if task.retry_count < task.max_retries:
                    print(f"[{datetime.now()}] 任务失败，重试 {task.retry_count}/{task.max_retries}")
                    await self.task_queue.put((-task.priority.value, task))
                else:
                    task.status = TaskStatus.FAILED
                    print(f"[{datetime.now()}] 任务最终失败: {task.id}")
                raise
    
    async def _invoke_model(self, task: FollowUpTask) -> Dict[str, Any]:
        """
        模拟o3-mini模型调用
        实际场景中会调用OpenAI API:
        POST https://api.openai.com/v1/responses
        """
        # 模拟API调用延迟
        await asyncio.sleep(0.5)
        return {
            "task_id": task.id,
            "status": "success",
            "output": f"Processed: {task.description[:50]}",
            "tokens_used": {
                "input": 1500,
                "output": 3200,
                "reasoning": 5000
            },
            "cost_estimate_usd": (
                1500 * 1.10 / 1_000_000 +  # 输入成本
                8200 * 4.40 / 1_000_000     # 输出+推理Token成本
            )
        }
    
    async def run_forever(self):
        """
        持续运行直到被外部休眠 - 持久模式的核心循环
        """
        self._running = True
        print(f"[{datetime.now()}] 持久化调度器启动，最大并发: {self.max_concurrent}")
        print("  等待任务中... (使用 sleep() 方法可休眠调度器)")
        
        while self._running:
            try:
                # 等待新任务，带超时以便检查休眠信号
                _, task = await asyncio.wait_for(
                    self.task_queue.get(), timeout=1.0
                )
                
                # 启动异步执行
                worker = asyncio.create_task(self.execute_task(task))
                self.active_tasks[task.id] = worker
                
                # 清理已完成的任务
                done_tasks = [
                    tid for tid, t in self.active_tasks.items()
                    if t.done()
                ]
                for tid in done_tasks:
                    self.active_tasks.pop(tid)
                    
            except asyncio.TimeoutError:
                continue
    
    def sleep(self):
        """休眠调度器 - 对应"put to sleep"指令"""
        self._running = False
        print(f"[{datetime.now()}] 调度器进入休眠状态")
        print(f"  活跃任务: {len(self.active_tasks)}")
        print(f"  完成任务: {len(self.completed_tasks)}")


# 使用示例
async def main():
    scheduler = PersistentScheduler(max_concurrent=2)
    
    # 模拟Codex在完成用户请求后自动创建后续任务
    await scheduler.create_follow_up(
        "检查代码库中未使用的导入并清理",
        priority=TaskPriority.LOW,
        context="已完成主功能重构，代码库需要清理"
    )
    
    await scheduler.create_follow_up(
        "为新增的API端点编写单元测试",
        priority=TaskPriority.HIGH,
        context="新增了3个REST API端点，覆盖率需达到90%"
    )
    
    # 启动调度器
    scheduler_task = asyncio.create_task(scheduler.run_forever())
    
    # 运行一段时间后休眠
    await asyncio.sleep(5)
    scheduler.sleep()
    await scheduler_task

if __name__ == "__main__":
    asyncio.run(main())

"""
Codex持久模式 - API调用示例
使用OpenAI Responses API的background模式
"""

from openai import OpenAI
import time
from typing import Optional, Dict, Any


class PersistentCodexClient:
    """
    持久模式Codex客户端
    通过background=True参数实现异步长时任务
    """
    
    def __init__(self, api_key: Optional[str] = None):
        self.client = OpenAI(api_key=api_key)
        self.active_tasks: Dict[str, str] = {}
    
    def start_persistent_task(
        self,
        prompt: str,
        reasoning_effort: str = "persistent",
        model: str = "o3-mini",
        context: Optional[Dict[str, Any]] = None
    ) -> str:
        """
        启动持久化任务
        
        使用background=True参数让任务在后台异步执行
        这是持久模式的关键API特征
        """
        messages = [
            {
                "role": "system",
                "content": (
                    "You are in persistent mode. Continue working until put to sleep. "
                    "After completing the user's request, proactively create follow-up tasks "
                    "and continue working across sessions. Use your knowledge of the user "
                    "to determine what to work on next."
                )
            }
        ]
        
        if context:
            messages.append({
                "role": "system",
                "content": f"Session context: {json.dumps(context)}"
            })
        
        messages.append({
            "role": "user",
            "content": prompt
        })
        
        # 发起异步后台请求
        response = self.client.responses.create(
            model=model,
            input=messages,
            background=True,  # 核心参数：启用后台异步执行
            reasoning={
                "effort": reasoning_effort
            },
            tools=[{
                "type": "function",
                "function": {
                    "name": "create_follow_up_task",
                    "description": "Create a follow-up task for later execution",
                    "parameters": {
                        "type": "object",
                        "properties": {
                            "description": {
                                "type": "string",
                                "description": "Task description"
                            },
                            "priority": {
                                "type": "string",
                                "enum": ["low", "medium", "high"]
                            }
                        },
                        "required": ["description"]
                    }
                }
            }]
        )
        
        task_id = response.id
        self.active_tasks[task_id] = "in_progress"
        print(f"持久化任务已启动: {task_id}")
        return task_id
    
    def poll_task_status(self, task_id: str) -> str:
        """
        轮询后台任务状态
        用于检查持久化任务的执行进度
        """
        response = self.client.responses.retrieve(task_id)
        status = response.status
        
        if status != self.active_tasks.get(task_id):
            print(f"任务 {task_id} 状态变更: {self.active_tasks.get(task_id)} -> {status}")
            self.active_tasks[task_id] = status
            
        return status
    
    def wait_for_completion(self, task_id: str, 
                            poll_interval: int = 30,
                            timeout: int = 3600) -> Dict[str, Any]:
        """
        等待任务完成（或超时）
        在持久模式下，任务可能持续数小时
        """
        start_time = time.time()
        while time.time() - start_time < timeout:
            status = self.poll_task_status(task_id)
            
            if status in ["completed", "failed"]:
                response = self.client.responses.retrieve(task_id)
                return {
                    "task_id": task_id,
                    "status": status,
                    "output": response.output_text,
                    "duration_minutes": (time.time() - start_time) / 60,
                    "usage": {
                        "input_tokens": response.usage.input_tokens,
                        "output_tokens": response.usage.output_tokens,
                        "total_tokens": response.usage.total_tokens
                    }
                }
            
            print(f"任务仍在执行中... (已运行 {(time.time() - start_time)/60:.1f} 分钟)")
            time.sleep(poll_interval)
        
        raise TimeoutError(f"任务 {task_id} 在 {timeout} 秒内未完成")


# 使用示例：夜间自动化构建
async def nightly_build_workflow():
    """
    典型场景：开发者在睡前启动一个持久化任务
    Codex会在夜间持续工作，自动完成代码审查、测试和修复
    """
    client = PersistentCodexClient()
    
    task_id = client.start_persistent_task(
        prompt=(
            "1. 审查项目中所有新提交的代码变更\n"
            "2. 运行完整的测试套件，记录所有失败的测试\n"
            "3. 自动修复失败的测试用例\n"
            "4. 更新API文档以匹配最新的代码变更\n"
            "5. 如果发现性能瓶颈，创建优化建议报告\n"
            "如果以上任何步骤遇到需要用户决策的问题，"
            "记录下来并在用户醒来时汇报"
        ),
        reasoning_effort="persistent",
        context={
            "project": "e-commerce-platform",
            "branch": "feature/payment-v2",
            "user_preferences": {
                "test_framework": "pytest",
                "coverage_threshold": 85,
                "style_guide": "PEP 8"
            }
        }
    )
    
    print(f"✅ 夜间自动化构建已启动: {task_id}")
    print("💤 开发者可以关闭电脑，Codex将在云端持续工作")
    print("📋 明早可通过 task_id 查询完整执行报告")
    
    return task_id

// Codex持久模式 - 跨会话任务状态管理
// 基于Go语言的线程状态持久化示例

package main

import (
	"context"
	"encoding/json"
	"fmt"
	"log"
	"os"
	"path/filepath"
	"sync"
	"time"
)

// ThreadState 表示一个Codex线程的持久化状态
type ThreadState struct {
	ID            string    `json:"id"`
	CreatedAt     time.Time `json:"created_at"`
	LastActiveAt  time.Time `json:"last_active_at"`
	Status        string    `json:"status"` // active, sleeping, completed
	ContextSummary string   `json:"context_summary"`
	FollowUpTasks []Task    `json:"follow_up_tasks"`
	ProjectPath   string    `json:"project_path"`
	UserProfile   string    `json:"user_profile_hash"`
	TokenUsage    Usage     `json:"token_usage"`
}

// Task 表示一个后续任务
type Task struct {
	ID          string    `json:"id"`
	Description string    `json:"description"`
	Status      string    `json:"status"` // pending, running, completed, failed
	Priority    int       `json:"priority"`
	CreatedAt   time.Time `json:"created_at"`
	ParentID    string    `json:"parent_id,omitempty"`
}

// Usage 记录Token消耗
type Usage struct {
	InputTokens   int `json:"input_tokens"`
	OutputTokens  int `json:"output_tokens"`
	ReasoningTokens int `json:"reasoning_tokens"`
}

// PersistentThreadManager 管理跨会话持久化线程
type PersistentThreadManager struct {
	mu       sync.RWMutex
	threads  map[string]*ThreadState
	storeDir string
}

// NewPersistentThreadManager 创建线程管理器
func NewPersistentThreadManager(storeDir string) *PersistentThreadManager {
	os.MkdirAll(storeDir, 0755)
	return &PersistentThreadManager{
		threads:  make(map[string]*ThreadState),
		storeDir: storeDir,
	}
}

// SaveThreadState 持久化线程状态到磁盘
func (m *PersistentThreadManager) SaveThreadState(state *ThreadState) error {
	m.mu.Lock()
	defer m.mu.Unlock()

	data, err := json.MarshalIndent(state, "", "  ")
	if err != nil {
		return fmt.Errorf("序列化线程状态失败: %w", err)
	}

	filePath := filepath.Join(m.storeDir, fmt.Sprintf("thread_%s.json", state.ID))
	if err := os.WriteFile(filePath, data, 0644); err != nil {
		return fmt.Errorf("写入线程状态文件失败: %w", err)
	}

	m.threads[state.ID] = state
	log.Printf("[持久化] 线程 %s 状态已保存 (任务数: %d)", 
		state.ID, len(state.FollowUpTasks))
	return nil
}

// ResumeThread 恢复之前休眠的线程
func (m *PersistentThreadManager) ResumeThread(threadID string) (*ThreadState, error) {
	// 先查内存
	if state, ok := m.threads[threadID]; ok {
		state.LastActiveAt = time.Now()
		state.Status = "active"
		log.Printf("[恢复] 从内存恢复线程 %s", threadID)
		return state, nil
	}

	// 从磁盘恢复
	filePath := filepath.Join(m.storeDir, fmt.Sprintf("thread_%s.json", threadID))
	data, err := os.ReadFile(filePath)
	if err != nil {
		if os.IsNotExist(err) {
			return nil, fmt.Errorf("线程 %s 不存在，无法恢复", threadID)
		}
		return nil, fmt.Errorf("读取线程状态失败: %w", err)
	}

	var state ThreadState
	if err := json.Unmarshal(data, &state); err != nil {
		return nil, fmt.Errorf("解析线程状态失败: %w", err)
	}

	state.LastActiveAt = time.Now()
	state.Status = "active"

	m.mu.Lock()
	m.threads[threadID] = &state
	m.mu.Unlock()

	log.Printf("[恢复] 从磁盘恢复线程 %s (上次活动: %s, 任务数: %d)",
		threadID, state.LastActiveAt.Format(time.RFC3339), len(state.FollowUpTasks))

	return &state, nil
}

// ListActiveTasks 列出所有活跃的后续任务
func (m *PersistentThreadManager) ListActiveTasks() []Task {
	m.mu.RLock()
	defer m.mu.RUnlock()

	var tasks []Task
	for _, thread := range m.threads {
		for _, task := range thread.FollowUpTasks {
			if task.Status == "pending" || task.Status == "running" {
				tasks = append(tasks, task)
			}
		}
	}
	return tasks
}

// PutThreadToSleep 休眠线程（对应"put to sleep"指令）
func (m *PersistentThreadManager) PutThreadToSleep(threadID string) error {
	m.mu.Lock()
	defer m.mu.Unlock()

	state, ok := m.threads[threadID]
	if !ok {
		return fmt.Errorf("线程 %s 未找到", threadID)
	}

	state.Status = "sleeping"
	state.LastActiveAt = time.Now()

	// 立即持久化
	if err := m.SaveThreadState(state); err != nil {
		return fmt.Errorf("休眠持久化失败: %w", err)
	}

	log.Printf("[休眠] 线程 %s 已休眠 (运行时长: %v)", 
		threadID, time.Since(state.CreatedAt))
	return nil
}

func main() {
	ctx := context.Background()
	manager := NewPersistentThreadManager("./thread_store")

	// 创建一个新的持久化线程
	thread := &ThreadState{
		ID:           "thread_alpha_001",
		CreatedAt:    time.Now(),
		LastActiveAt: time.Now(),
		Status:       "active",
		ProjectPath:  "/workspace/backend-api",
		FollowUpTasks: []Task{
			{
				ID:          "task_001",
				Description: "为用户认证模块添加单元测试覆盖",
				Status:      "completed",
				Priority:    2,
				CreatedAt:   time.Now().Add(-2 * time.Hour),
			},
			{
				ID:          "task_002",
				Description: "重构数据库连接池配置以提升性能",
				Status:      "pending",
				Priority:    1,
				CreatedAt:   time.Now().Add(-1 * time.Hour),
				ParentID:    "task_001",
			},
			{
				ID:          "task_003",
				Description: "更新API文档以反映最新的端点变更",
				Status:      "pending",
				Priority:    0,
				CreatedAt:   time.Now(),
				ParentID:    "task_002",
			},
		},
		TokenUsage: Usage{
			InputTokens:     1250000,
			OutputTokens:    380000,
			ReasoningTokens: 920000,
		},
	}

	// 保存线程状态
	if err := manager.SaveThreadState(thread); err != nil {
		log.Fatalf("保存线程状态失败: %v", err)
	}

	// 模拟跨会话恢复
	fmt.Println("=== 模拟跨会话恢复 ===")
	fmt.Println("用户关闭了终端，Codex会话结束...")
	fmt.Println("12小时后，用户重新打开终端...")

	time.Sleep(1 * time.Second) // 模拟时间流逝

	resumed, err := manager.ResumeThread("thread_alpha_001")
	if err != nil {
		log.Fatalf("恢复线程失败: %v", err)
	}

	fmt.Printf("✅ 线程恢复成功: %s\n", resumed.ID)
	fmt.Printf("📋 项目路径: %s\n", resumed.ProjectPath)
	fmt.Printf("🔄 待处理任务: %d\n", len(manager.ListActiveTasks()))
	fmt.Printf("💰 Token消耗: 输入=%d, 输出=%d, 推理=%d\n",
		resumed.TokenUsage.InputTokens,
		resumed.TokenUsage.OutputTokens,
		resumed.TokenUsage.ReasoningTokens)
	fmt.Printf("💤 休眠2小时后自动恢复执行...\n")

	// 休眠线程
	if err := manager.PutThreadToSleep("thread_alpha_001"); err != nil {
		log.Fatalf("休眠线程失败: %v", err)
	}
	_ = ctx
}

"""
持久模式成本估算模型
"""

def estimate_persistent_cost(
    hours: float,
    avg_input_tokens_per_hour: int = 50000,
    avg_output_tokens_per_hour: int = 30000,
    reasoning_tokens_multiplier: float = 3.0,
    input_price: float = 1.10,   # $/M tokens
    output_price: float = 4.40,  # $/M tokens
    cache_hit_rate: float = 0.3  # 30%缓存命中率
) -> dict:
    """
    估算持久模式运行成本
    
    Parameters:
    -----------
    hours : float
        持续运行小时数
    avg_input_tokens_per_hour : int
        每小时平均输入Token数
    avg_output_tokens_per_hour : int
        每小时平均输出Token数
    reasoning_tokens_multiplier : float
        推理Token倍率（o3-mini的隐藏推理Token）
    cache_hit_rate : float
        输入缓存命中率
    
    Returns:
    --------
    dict : 成本明细
    """
    # 总Token数
    total_input = avg_input_tokens_per_hour * hours
    # 推理Token计入输出（隐藏推理Token）
    total_output = avg_output_tokens_per_hour * hours * reasoning_tokens_multiplier
    
    # 缓存命中节省
    cache_hit_input = total_input * cache_hit_rate
    cache_miss_input = total_input * (1 - cache_hit_rate)
    
    # 成本计算
    input_cost = (cache_miss_input * input_price + 
                  cache_hit_input * 0.275) / 1_000_000
    output_cost = total_output * output_price / 1_000_000
    
    total_cost = input_cost + output_cost
    
    return {
        "运行时长": f"{hours:.1f}小时",
        "总输入Token": f"{total_input:,}",
        "总输出Token(含推理)": f"{int(total_output):,}",
        "输入成本": f"${input_cost:.2f}",
        "输出成本": f"${output_cost:.2f}",
        "总成本": f"${total_cost:.2f}",
        "每小时平均成本": f"${total_cost/hours:.2f}"
    }

# 典型场景成本估算
scenarios = [
    ("夜间自动化构建 (8小时)", 8),
    ("全日代码审查 (24小时)", 24),
    ("周末持续运行 (48小时)", 48),
    ("最长连续运行 (25小时)", 25)
]

print("=" * 60)
print("Codex持久模式 - 成本估算")
print("=" * 60)
print(f"模型: o3-mini (输入: $1.10/M, 输出: $4.40/M)")
print(f"推理Token倍率: 3x (含隐藏推理Token)")
print()

for name, hours in scenarios:
    cost = estimate_persistent_cost(hours)
    print(f"┌─ {name} ─────────────────────────────┐")
    for k, v in cost.items():
        print(f"│ {k}: {v}")
    print("└──────────────────────────────────────────┘")
    print()

# 输出示例:
# ┌─ 夜间自动化构建 (8小时) ─────────────────────────────┐
# │ 运行时长: 8.0小时
# │ 总输入Token: 400,000
# │ 总输出Token(含推理): 720,000
# │ 输入成本: $0.40
# │ 输出成本: $3.17
# │ 总成本: $3.57
# │ 每小时平均成本: $0.45
# └──────────────────────────────────────────┘

"""
独立开发者夜间工作流
"""

from codex_sdk import Codex, CodexOptions
import asyncio

async def nightly_workflow():
    codex = Codex(
        CodexOptions(
            config_overrides={
                "reasoning_effort": "persistent",
                "max_runtime_hours": 8,
                "auto_approve_patches": False,
                "notification_on_complete": True,
            }
        )
    )
    
    # 创建持久化线程
    thread = codex.start_thread()
    
    # 分配夜间任务
    await thread.run(
        "完整分析项目代码库:\n"
        "1. 识别所有已废弃的API端点并标记\n"
        "2. 运行完整测试套件，修复所有失败的测试\n"
        "3. 检查所有第三方依赖的最新版本，评估升级影响\n"
        "4. 生成一份完整的代码质量报告\n"
        "5. 如果发现安全漏洞，立即暂停并通知我",
        reasoning_effort="persistent"
    )
    
    print("✅ 夜间任务已启动，明早查看结果")
    return thread.id

# 运行
asyncio.run(nightly_workflow())