Cursor IPO：AI编程赛道的万亿级估值里程碑

class CodebaseIndexer:
    """
    维护整个代码库的语义索引，
    实现上下文感知的AI辅助。
    """
    
    def __init__(self, project_root: Path):
        self.project_root = project_root
        self.index = SemanticIndex()
        self.file_watcher = FileWatcher(project_root)
        self.ast_parser = ASTParser()
        
    def index_project(self) -> None:
        """启动时对整个项目建立索引。"""
        for file_path in self.project_root.rglob("*.py"):
            if self._should_index(file_path):
                self._index_file(file_path)
                
    def _index_file(self, file_path: Path) -> None:
        """解析并索引单个文件。"""
        tree = self.ast_parser.parse(file_path)
        symbols = self._extract_symbols(tree)
        imports = self._extract_imports(tree)
        
        self.index.add(
            file_path=file_path,
            symbols=symbols,
            imports=imports,
            content_hash=hash(file_path.read_text())
        )
        
    def get_context(self, cursor_position: int, file_path: Path) -> CodeContext:
        """获取AI建议的相关上下文。"""
        current_file = self.index.get(file_path)
        related_files = self.index.find_related(
            imports=current_file.imports,
            symbols=current_file.symbols
        )
        return CodeContext(
            current_file=current_file,
            related_files=related_files,
            cursor_position=cursor_position
        )

from enum import Enum
from dataclasses import dataclass
from typing import Optional

class Model(Enum):
    COMPOSER_25 = "composer-2.5"
    CLAUDE_OPUS = "claude-opus-4.6"
    GPT_54 = "gpt-5.4"
    GEMINI = "gemini-pro"

@dataclass
class ModelConfig:
    name: Model
    context_window: int
    cost_per_1k_input: float
    cost_per_1k_output: float
    best_for: list[str]

MODEL_CONFIGS = {
    Model.COMPOSER_25: ModelConfig(
        name=Model.COMPOSER_25,
        context_window=128_000,
        cost_per_1k_input=0.0005,  # 仅$0.5/百万token！
        cost_per_1k_output=0.0025,
        best_for=["快速补全", "重构", "简单编辑"]
    ),
    Model.CLAUDE_OPUS: ModelConfig(
        name=Model.CLAUDE_OPUS,
        context_window=1_000_000,  # 100万token上下文！
        cost_per_1k_input=0.015,
        cost_per_1k_output=0.075,
        best_for=["复杂推理", "大型重构", "调试"]
    ),
}

class ModelRouter:
    """基于任务特征智能选择最优模型。"""
    
    def __init__(self):
        self.models = MODEL_CONFIGS
        self.usage_tracker = UsageTracker()
        
    def route(self, task: Task) -> Model:
        """根据任务特征选择最佳模型。"""
        # 快速补全 → Composer（便宜50倍）
        if task.type in ["completion", "inline_edit"]:
            if task.complexity == "low":
                return Model.COMPOSER_25
        
        # 复杂推理 → Claude Opus（100万token上下文）
        if task.complexity == "high" or task.context_size > 500_000:
            return Model.CLAUDE_OPUS
        
        # 默认为Auto模式
        return Model.COMPOSER_25  # 成本效益默认选择

# Cursor如何使用Git Worktrees实现安全的智能体执行

# 智能体收到任务："将认证逻辑提取到hooks"
git worktree add /tmp/cursor-agent-workspace-{uuid}

# 智能体在隔离分支中工作
cd /tmp/cursor-agent-workspace-{uuid}
git checkout -b cursor-agent-task-{task_id}

# 进行更改...
git add . && git commit -m "Extract auth logic"

# 生成diff供用户审核
git diff main...cursor-agent-task-{task_id}

# 审核后清理
git worktree remove /tmp/cursor-agent-workspace-{uuid}

class ComposerModel:
    """
    Cursor的Composer 2.5 - 以极低成本提供接近前沿水平的编程模型。
    """
    
    def __init__(self):
        self.token_speed = 200  # token/秒
        self.latency_target = 30  # 秒
        self.parallel_capacity = 8  # 并发智能体数
        
    def generate_code(self, prompt: str, context: CodeContext):
        """
        主代码生成流程：
        1. 深度上下文理解
        2. 意图识别
        3. 快速代码生成
        4. 自我审查与验证
        """
        # 阶段1：上下文理解
        understanding = self.analyze_context(context)
        
        # 阶段2：意图识别
        intent = self.identify_intent(prompt, understanding)
        
        # 阶段3：代码生成（200 tok/s！）
        code = self.generate_fast(intent, understanding)
        
        # 阶段4：自我审查
        reviewed_code = self.self_review(code, context)
        
        return reviewed_code
    
    def self_review(self, code: str, context: CodeContext):
        """返回前自动进行代码审查。"""
        checks = {
            'syntax': self.check_syntax(code),
            'style': self.check_style(code, context.style_guide),
            'security': self.check_security(code),
            'imports': self.verify_imports(code, context)
        }
        
        if all(checks.values()):
            return code
        return self.refine(code, checks)

from dataclasses import dataclass
from typing import List, Protocol
import asyncio

@dataclass
class Subtask:
    description: str
    agent_type: str
    priority: int

class Agent(Protocol):
    async def execute(self, task: Subtask, worktree: str) -> AgentResult: ...

class MultiAgentOrchestrator:
    """
    协调多个AI智能体完成复杂任务。
    关键创新：智能体在git worktree中并行工作。
    """
    
    def __init__(self):
        self.agents: dict[str, Agent] = {}
        self.worktree_manager = GitWorktreeManager()
        
    async def execute_complex_task(self, task: ComplexTask) -> UnifiedResult:
        # 1. 任务分解
        subtasks = await self.decompose_task(task)
        
        # 2. 智能体分配
        assignments = self.assign_agents(subtasks)
        
        # 3. 并行执行（最多8个智能体！）
        worktrees = await self.worktree_manager.create_multiple(len(assignments))
        
        results = await asyncio.gather(*[
            self._execute_with_isolation(agent, subtask, wt)
            for (agent, subtask), wt in zip(assignments, worktrees)
        ])
        
        # 4. 结果合并
        return self.merge_results(results)
    
    async def _execute_with_isolation(
        self, agent: Agent, subtask: Subtask, worktree: str
    ) -> AgentResult:
        """在隔离的git worktree中执行智能体。"""
        try:
            result = await agent.execute(subtask, worktree)
            await self.worktree_manager.cleanup(worktree)
            return result
        except Exception as e:
            await self.worktree_manager.cleanup(worktree)  # 始终清理
            raise

# Claude Opus 4.8 技术规格

OPUS_48_SPECS = {
    "parameters": "660亿",  # 从460亿增加
    "context_window": 1_000_000,  # 100万token！
    "training_data": "增强的编程语料库",
    "specializations": [
        "动态工作流",
        "多步骤推理", 
        "代码生成",
        "调试与修复"
    ],
    "pricing": {
        "每百万输入token": "$15.00",
        "每百万输出token": "$75.00"
    }
}

class DynamicWorkflow:
    """
    Opus 4.8的突破：自主动态工作流。
    模型可以在没有持续人工指导的情况下规划、执行和迭代。
    """
    
    async def execute(self, task: str, context: Codebase):
        # 1. 规划：创建执行路线图
        plan = await self.planner.create(task, context)
        
        # 2. 执行：运行计划步骤
        for step in plan.steps:
            result = await step.execute(context)
            
            # 3. 验证：检查结果
            if not step.validate(result):
                # 4. 迭代：自我修正
                step = await self.iterate(step, result)
        
        return plan.final_result

# OpenAI Codex 2026年6月：Sites和Plugins

class CodexPlatform:
    """
    Codex已从编程工具扩展到通用知识工作领域。
    """
    
    # 2026年6月新功能
    NEW_FEATURES = {
        "sites": {
            "description": "创建交互式托管Web应用",
            "用例": [
                "仪表盘",
                "计划器", 
                "审核工作区",
                "项目看板"
            ],
            "目标用户": "Business和Enterprise"
        },
        "annotations": {
            "description": "任意内容原地编辑",
            "支持": [
                "代码",
                "文档",
                "幻灯片",
                "电子表格"
            ]
        },
        "plugins": {
            "数量": 6,
            "分类": [
                "数据分析",      # Snowflake, Tableau, Databricks
                "创意制作",     # Figma, Canva, Shutterstock
                "销售",         # Salesforce, HubSpot, Slack
                "产品设计",     # Figma, Canva
                "公募投资",     # S&P, PitchBook, Moody's
                "投资银行"      # 尽职调查、可比公司分析
            ],
            "集成应用数": 62,
            "自动化技能数": 110
        }
    }
    
    @property
    def non_developer_adoption(self):
        """
        关键指标：非开发者采用率
        """
        return {
            "用户占比": "20%",  # 从2025年的约5%上升
            "采用速度": "是工程师的3倍",
            "趋势": "加速增长"
        }

# 前：手动重构600行React组件
# 手动操作需要约45分钟

# 后：使用Cursor Composer
# 时间：约6分钟，包括手动清理

# 示例：将认证逻辑提取到自定义hooks

# 原始单体组件（简化版）
class UserProfileComponent(React.Component):
    def componentDid_mount(self):
        token = localStorage.get_item('auth_token')
        if not token:
            self.props.router.navigate('/login')
            return
        
        user_id = self.parse_jwt(token).user_id
        self.fetch_user_data(user_id)
    
    def handle_login(self, credentials):
        auth_token = self.authenticate(credentials)
        localStorage.setItem('auth_token', auth_token)
        self.fetch_user_data(credentials.user_id)
    
    def handle_logout(self):
        localStorage.removeItem('auth_token')
        self.props.router.navigate('/')

# Cursor Composer提示：
# "将认证逻辑提取到useAuth hook、
#  useToken管理hook，并更新所有5个
#  当前内联管理认证状态的组件"

# 结果：Composer生成以下文件：

# hooks/useAuth.ts
import { useState, useEffect, useCallback } from 'react';
import { useAuthToken } from './useAuthToken';
import { useUser } from './useUser';
import { authenticate, AuthCredentials, User } from '@/api/auth';

interface UseAuthReturn {
  user: User | null;
  isAuthenticated: boolean;
  isLoading: boolean;
  login: (credentials: AuthCredentials) => Promise<void>;
  logout: () => void;
  error: Error | null;
}

export function useAuth(): UseAuthReturn {
  const { token, setToken, clearToken } = useAuthToken();
  const { user, fetchUser, clearUser } = useUser();
  
  const [isLoading, setIsLoading] = useState(false);
  const [error, setError] = useState<Error | null>(null);
  
  const isAuthenticated = !!token && !!user;
  
  const login = useCallback(async (credentials: AuthCredentials) => {
    setIsLoading(true);
    setError(null);
    try {
      const { token: newToken, user: newUser } = await authenticate(credentials);
      setToken(newToken);
      await fetchUser(newUser.id);
    } catch (err) {
      setError(err as Error);
      throw err;
    } finally {
      setIsLoading(false);
    }
  }, [setToken, fetchUser]);
  
  const logout = useCallback(() => {
    clearToken();
    clearUser();
  }, [clearToken, clearUser]);
  
  // 挂载时自动获取用户（如果token存在）
  useEffect(() => {
    if (token && !user) {
      fetchUser();
    }
  }, [token, user, fetchUser]);
  
  return { user, isAuthenticated, isLoading, login, logout, error };
}

# hooks/useAuthToken.ts
import { useState, useCallback } from 'react';

const TOKEN_KEY = 'auth_token';

interface UseAuthTokenReturn {
  token: string | null;
  setToken: (token: string) => void;
  clearToken: () => void;
  parseToken: () => { user_id: string } | null;
}

export function useAuthToken(): UseAuthTokenReturn {
  const [token, setTokenState] = useState<string | null>(
    () => localStorage.getItem(TOKEN_KEY)
  );
  
  const setToken = useCallback((newToken: string) => {
    localStorage.setItem(TOKEN_KEY, newToken);
    setTokenState(newToken);
  }, []);
  
  const clearToken = useCallback(() => {
    localStorage.removeItem(TOKEN_KEY);
    setTokenState(null);
  }, []);
  
  const parseToken = useCallback((): { user_id: string } | null => {
    if (!token) return null;
    try {
      const payload = JSON.parse(atob(token.split('.')[1]));
      return payload;
    } catch {
      return null;
    }
  }, [token]);
  
  return { token, setToken, clearToken, parseToken };
}

# Updated UserProfile.tsx - 现在使用hooks！
import { useAuth } from '@/hooks/useAuth';

export function UserProfile() {
  const { user, isAuthenticated, logout } = useAuth();
  
  if (!isAuthenticated) {
    return <Navigate to="/login" />;
  }
  
  return (
    <div className="user-profile">
      <h1>欢迎，{user.name}</h1>
      <button onClick={logout}>退出登录</button>
    </div>
  );
}

# Cursor智能体工作流：调查生产环境支付webhook超时问题

# 用户提示：
# "调查为什么支付webhook间歇性超时。
#  检查最近24小时的日志，
#  识别根本原因，并提出修复方案。"

# 智能体执行跟踪：

"""
[智能体] 开始调查...
[智能体] 查询最近24小时payment-webhook日志...
[智能体] 发现10,000个请求中有147个超时错误
[智能体] 分析错误模式...

分析结果：
├── 错误类型："30秒后连接超时"
├── 模式：部署后15-45分钟出现
├── 相关性：观察到内存使用量激增
├── 根本原因：数据库连接池耗尽
│   └── 超时时连接未正确释放
└── 建议修复：在finally块中添加正确清理

[智能体] 生成修复代码...
"""

# 生成的修复代码：
import asyncpg
from contextlib import asynccontextmanager
from typing import AsyncGenerator

class PaymentDatabase:
    """具有正确连接处理的数据库管理器。"""
    
    def __init__(self, pool_size: int = 20):
        self.pool: asyncpg.Pool | None = None
        self.pool_size = pool_size
    
    async def connect(self):
        self.pool = await asyncpg.create_pool(
            min_size=5,
            max_size=self.pool_size,
            command_timeout=30
        )
    
    async def close(self):
        if self.pool:
            await self.pool.close()
    
    @asynccontextmanager
    async def acquire(self) -> AsyncGenerator[asyncpg.Connection, None]:
        """
        安全连接获取的上下文管理器。
        关键修复：即使超时也始终释放连接。
        """
        if not self.pool:
            raise RuntimeError("数据库未连接")
        
        connection = await self.pool.acquire()
        try:
            yield connection
        except asyncio.TimeoutError:
            # 记录超时供监控
            logger.warning("连接超时，释放回连接池")
            raise
        finally:
            # 关键：即使超时也释放连接
            await self.pool.release(connection)
    
    async def execute_payment_webhook(
        self, 
        transaction_id: str,
        timeout: float = 30.0
    ) -> dict:
        """
        执行具有正确超时处理的支付webhook。
        """
        async with self.acquire() as conn:
            # 设置语句超时（防止长时间运行查询）
            await conn.execute(
                f"SET statement_timeout = '{int(timeout * 1000)}ms'"
            )
            
            result = await conn.fetchrow(
                """
                SELECT * FROM payment_transactions 
                WHERE transaction_id = $1
                """,
                transaction_id
            )
            
            # 重置后续操作的超时
            await conn.execute("RESET statement_timeout")
            
            return dict(result) if result else None

# 测试用例验证修复
import pytest
from unittest.mock import AsyncMock, patch
import asyncio

@pytest.mark.asyncio
async def test_timeout时连接正确释放():
    """
    验证即使超时也正确释放连接。
    此测试在修复前会失败。
    """
    db = PaymentDatabase(pool_size=2)
    await db.connect()
    
    # 模拟超时场景
    with patch.object(db.pool, 'acquire', new_callable=AsyncMock) as mock_acquire:
        mock_conn = AsyncMock()
        mock_acquire.return_value = mock_conn
        
        # 强制超时
        with pytest.raises(asyncio.TimeoutError):
            async with db.acquire() as conn:
                await asyncio.sleep(35)  # 超过超时时间
        
        # 关键断言：即使超时也释放连接
        mock_conn.reconnect.assert_not_called()
        mock_conn.release.assert_called_once()  # 已修复！
    
    await db.close()

# Cursor可以从规格说明生成完整的API端点

# 用户规格：
"""
为待办事项列表创建REST API，包含：
- GET /todos - 列出所有待办（分页）
- POST /todos - 创建新待办
- GET /todos/{id} - 获取特定待办
- PUT /todos/{id} - 更新待办
- DELETE /todos/{id} - 删除待办

每个待办事项包含：
- id (UUID)
- title (字符串，必填)
- description (字符串，可选)
- completed (布尔值，默认false)
- created_at (时间戳)
- updated_at (时间戳)

使用FastAPI、SQLAlchemy和PostgreSQL。
包含适当的验证、错误处理和OpenAPI文档。
"""

# Cursor生成完整实现：

# models/todo.py
from datetime import datetime
from uuid import UUID, uuid4
from sqlalchemy import Column, String, Boolean, DateTime
from sqlalchemy.dialects.postgresql import UUID as PGUUID
from sqlalchemy.orm import DeclarativeBase

class Base(DeclarativeBase):
    pass

class Todo(Base):
    __tablename__ = "todos"
    
    id: Mapped[UUID] = Column(
        PGUUID(as_uuid=True),
        primary_key=True,
        default=uuid4
    )
    title: Mapped[str] = Column(
        String(255),
        nullable=False,
        index=True
    )
    description: Mapped[str | None] = Column(
        String(2000),
        nullable=True
    )
    completed: Mapped[bool] = Column(
        Boolean,
        default=False,
        nullable=False
    )
    created_at: Mapped[datetime] = Column(
        DateTime,
        default=datetime.utcnow,
        nullable=False
    )
    updated_at: Mapped[datetime] = Column(
        DateTime,
        default=datetime.utcnow,
        onupdate=datetime.utcnow,
        nullable=False
    )
    
    def to_dict(self) -> dict:
        return {
            "id": str(self.id),
            "title": self.title,
            "description": self.description,
            "completed": self.completed,
            "created_at": self.created_at.isoformat(),
            "updated_at": self.updated_at.isoformat()
        }

# schemas/todo.py
from pydantic import BaseModel, Field, ConfigDict
from uuid import UUID
from datetime import datetime
from typing import Optional

class TodoBase(BaseModel):
    title: str = Field(
        ...,
        min_length=1,
        max_length=255,
        description="待办事项标题"
    )
    description: Optional[str] = Field(
        None,
        max_length=2000,
        description="可选的待办事项描述"
    )

class TodoCreate(TodoBase):
    pass

class TodoUpdate(BaseModel):
    title: Optional[str] = Field(None, min_length=1, max_length=255)
    description: Optional[str] = Field(None, max_length=2000)
    completed: Optional[bool] = None

class TodoResponse(TodoBase):
    model_config = ConfigDict(from_attributes=True)
    
    id: UUID
    completed: bool
    created_at: datetime
    updated_at: datetime

class PaginatedTodos(BaseModel):
    items: list[TodoResponse]
    total: int
    page: int
    page_size: int
    pages: int

# repositories/todo.py
from sqlalchemy import select, func
from sqlalchemy.ext.asyncio import AsyncSession
from uuid import UUID
from typing import Optional
from .models import Todo

class TodoRepository:
    def __init__(self, session: AsyncSession):
        self.session = session
    
    async def create(self, todo: Todo) -> Todo:
        self.session.add(todo)
        await self.session.commit()
        await self.session.refresh(todo)
        return todo
    
    async def get_by_id(self, todo_id: UUID) -> Optional[Todo]:
        result = await self.session.execute(
            select(Todo).where(Todo.id == todo_id)
        )
        return result.scalar_one_or_none()
    
    async def list(
        self, 
        page: int = 1, 
        page_size: int = 20
    ) -> tuple[list[Todo], int]:
        # 计数总数
        count_result = await self.session.execute(
            select(func.count()).select_from(Todo)
        )
        total = count_result.scalar_one()
        
        # 获取分页结果
        offset = (page - 1) * page_size
        result = await self.session.execute(
            select(Todo)
            .order_by(Todo.created_at.desc())
            .offset(offset)
            .limit(page_size)
        )
        todos = list(result.scalars().all())
        
        return todos, total
    
    async def update(self, todo: Todo) -> Todo:
        await self.session.commit()
        await self.session.refresh(todo)
        return todo
    
    async def delete(self, todo: Todo) -> None:
        await self.session.delete(todo)
        await self.session.commit()

# routers/todo.py
from fastapi import APIRouter, Depends, HTTPException, Query
from sqlalchemy.ext.asyncio import AsyncSession
from uuid import UUID
from typing import Optional
from ..database import get_db
from ..services.todo import TodoService
from ..repositories.todo import TodoRepository
from ..schemas.todo import (
    TodoCreate,
    TodoUpdate,
    TodoResponse,
    PaginatedTodos
)

router = APIRouter(prefix="/todos", tags=["待办事项"])

def get_todo_service(
    db: AsyncSession = Depends(get_db)
) -> TodoService:
    return TodoService(TodoRepository(db))

@router.post("/", response_model=TodoResponse, status_code=201)
async def create_todo(
    todo_data: TodoCreate,
    service: TodoService = Depends(get_todo_service)
):
    todo = await service.create_todo(
        title=todo_data.title,
        description=todo_data.description
    )
    return todo

@router.get("/", response_model=PaginatedTodos)
async def list_todos(
    page: int = Query(1, ge=1),
    page_size: int = Query(20, ge=1, le=100),
    service: TodoService = Depends(get_todo_service)
):
    result = await service.list_todos(page, page_size)
    return {
        **result,
        "items": [t.to_dict() for t in result["items"]]
    }

@router.get("/{todo_id}", response_model=TodoResponse)
async def get_todo(
    todo_id: UUID,
    service: TodoService = Depends(get_todo_service)
):
    todo = await service.get_todo(todo_id)
    if not todo:
        raise HTTPException(status_code=404, detail="待办事项不存在")
    return todo

@router.put("/{todo_id}", response_model=TodoResponse)
async def update_todo(
    todo_id: UUID,
    todo_data: TodoUpdate,
    service: TodoService = Depends(get_todo_service)
):
    todo = await service.update_todo(
        todo_id=todo_id,
        **todo_data.model_dump(exclude_unset=True)
    )
    if not todo:
        raise HTTPException(status_code=404, detail="待办事项不存在")
    return todo

@router.delete("/{todo_id}", status_code=204)
async def delete_todo(
    todo_id: UUID,
    service: TodoService = Depends(get_todo_service)
):
    deleted = await service.delete_todo(todo_id)
    if not deleted:
        raise HTTPException(status_code=404, detail="待办事项不存在")