Origin发布：Cursor如何用AI Agent时代的Git托管挑战GitHub霸权

#!/usr/bin/env python3
"""
示例：使用Cursor Agent API自动化Git操作
"""
import os
import subprocess
from typing import List, Optional

class OriginAgent:
    """Origin平台上的AI Agent操作封装"""
    
    def __init__(self, repo_url: str, token: str):
        self.repo_url = repo_url
        self.token = token
        self.workspace = f"/tmp/workspace/{os.urandom(4).hex()}"
    
    def clone(self, branch: str = "main") -> None:
        """克隆Origin仓库"""
        url = self.repo_url.replace("https://", f"https://oauth2:{self.token}@")
        subprocess.run(
            ["git", "clone", "--branch", branch, url, self.workspace],
            check=True, capture_output=True
        )
        print(f"✓ 已克隆仓库到 {self.workspace}")
    
    def create_branch(self, branch_name: str, base: str = "main") -> None:
        """创建特性分支"""
        subprocess.run(["git", "checkout", "-b", branch_name, base],
                       cwd=self.workspace, check=True)
        print(f"✓ 已创建分支 {branch_name} (基于 {base})")
    
    def modify_file(self, filepath: str, new_content: str) -> None:
        """修改文件内容"""
        full_path = os.path.join(self.workspace, filepath)
        os.makedirs(os.path.dirname(full_path), exist_ok=True)
        with open(full_path, 'w') as f:
            f.write(new_content)
        print(f"✓ 已修改文件 {filepath}")
    
    def commit(self, message: str) -> str:
        """提交变更"""
        subprocess.run(["git", "add", "-A"], cwd=self.workspace, check=True)
        result = subprocess.run(
            ["git", "commit", "-m", message],
            cwd=self.workspace, check=True, capture_output=True, text=True
        )
        commit_hash = result.stdout.split()[1] if result.stdout else "unknown"
        print(f"✓ 已提交: {commit_hash}")
        return commit_hash
    
    def push(self, branch: str) -> None:
        """推送分支到Origin"""
        subprocess.run(
            ["git", "push", "origin", branch],
            cwd=self.workspace, check=True
        )
        print(f"✓ 已推送分支 {branch} 到 Origin")
    
    def open_pr(self, title: str, body: str, head: str, base: str = "main") -> dict:
        """通过Origin API打开Pull Request"""
        import requests
        
        api_url = f"https://api.cursor.com/v1/codebase/repos/{self._repo_name()}/pulls"
        response = requests.post(
            api_url,
            headers={"Authorization": f"Bearer {self.token}"},
            json={
                "title": title,
                "body": body,
                "head": head,
                "base": base
            }
        )
        response.raise_for_status()
        pr_data = response.json()
        print(f"✓ 已创建 PR #{pr_data['number']}: {title}")
        return pr_data
    
    def _repo_name(self) -> str:
        """从URL提取仓库名"""
        return self.repo_url.rstrip('/').split('/')[-1]
    
    def cleanup(self) -> None:
        """清理工作区"""
        import shutil
        if os.path.exists(self.workspace):
            shutil.rmtree(self.workspace)
            print(f"✓ 已清理工作区")


# 使用示例：Agent自动修复Bug
agent = OriginAgent(
    repo_url="https://cursor.com/codebase/acme-corp/backend-api",
    token=os.environ["ORIGIN_TOKEN"]
)

try:
    # 1. 克隆仓库
    agent.clone()
    
    # 2. 创建修复分支
    agent.create_branch("fix/auth-timeout", "main")
    
    # 3. 修复代码
    fixed_code = """def authenticate(token: str, timeout: int = 30) -> dict:
    \"\"\"验证用户令牌，支持超时控制\"\"\"
    import time
    start = time.time()
    
    # 修复：增加超时检查和重试机制
    while time.time() - start < timeout:
        result = verify_token(token)
        if result["status"] == "success":
            return result
        if result["status"] == "retryable":
            time.sleep(0.5)
            continue
        break
    
    return {"status": "error", "message": "Authentication timeout"}
"""
    agent.modify_file("auth/service.py", fixed_code)
    
    # 4. 提交并推送
    commit_hash = agent.commit("fix: add timeout control to auth service")
    agent.push("fix/auth-timeout")
    
    # 5. 创建PR
    pr = agent.open_pr(
        title="fix: add timeout control to auth service",
        body=f"## 变更说明\n\n修复认证服务在负载过高时的超时问题。\n\n### 变更内容\n- 增加超时参数控制\n- 添加自动重试机制\n- 提交: {commit_hash}",
        head="fix/auth-timeout"
    )
    
finally:
    agent.cleanup()

// 示例：Origin的语义冲突解决算法（伪代码）
package merge

import (
	"context"
	"fmt"
	"strings"
)

// ConflictRegion 表示一个冲突区域
type ConflictRegion struct {
	FilePath    string
	Ours        string // 当前分支的代码
	Theirs      string // 目标分支的代码
	Base        string // 共同祖先的代码
	StartLine   int
	EndLine     int
}

// MergeResult 合并结果
type MergeResult struct {
	Resolved   bool
	Content    string
	Confidence float64 // 置信度 0.0-1.0
	Strategy   string  // 使用的策略
}

// SemanticMergeEngine 语义合并引擎
type SemanticMergeEngine struct {
	modelEndpoint string
}

// ResolveConflicts 自动解决合并冲突
func (e *SemanticMergeEngine) ResolveConflicts(
	ctx context.Context,
	conflicts []ConflictRegion,
) ([]MergeResult, error) {
	results := make([]MergeResult, 0, len(conflicts))
	
	for _, conflict := range conflicts {
		result, err := e.resolveSingle(ctx, conflict)
		if err != nil {
			// 降级到文本合并
			result = e.fallbackTextMerge(conflict)
		}
		results = append(results, result)
	}
	
	return results, nil
}

func (e *SemanticMergeEngine) resolveSingle(
	ctx context.Context,
	region ConflictRegion,
) (MergeResult, error) {
	// 策略1: 检测是否只是添加了不重叠的代码
	if e.isAdditiveChange(region.Base, region.Ours, region.Theirs) {
		return MergeResult{
			Resolved:   true,
			Content:    e.combineAdditive(region.Ours, region.Theirs),
			Confidence: 0.95,
			Strategy:   "additive",
		}, nil
	}
	
	// 策略2: 检测是否一方重命名了变量/函数
	if e.isRenameOnly(region.Base, region.Ours, region.Theirs) {
		return MergeResult{
			Resolved:   true,
			Content:    e.applyRename(region),
			Confidence: 0.90,
			Strategy:   "rename",
		}, nil
	}
	
	// 策略3: 使用AI模型进行语义合并
	aiResult, err := e.aiSemanticMerge(ctx, region)
	if err == nil && aiResult.Confidence > 0.8 {
		return aiResult, nil
	}
	
	// 策略4: 标记为需要人工介入
	return MergeResult{
		Resolved:   false,
		Content:    e.markForHumanReview(region),
		Confidence: 0.0,
		Strategy:   "human_review_required",
	}, nil
}

func (e *SemanticMergeEngine) isAdditiveChange(
	base, ours, theirs string,
) bool {
	// 检查两方的修改是否添加了不同的函数/方法
	ourAdditions := extractAdditions(base, ours)
	theirAdditions := extractAdditions(base, theirs)
	
	// 如果没有重叠的添加，则是加法变更
	for _, ourAdd := range ourAdditions {
		for _, theirAdd := range theirAdditions {
			if overlap(ourAdd, theirAdd) {
				return false
			}
		}
	}
	return true
}

// extractAdditions 提取从base到target新增的代码块
func extractAdditions(base, target string) []string {
	// 使用AST对比提取新增的函数/方法
	// 实际实现中会调用Tree-sitter或类似工具
	return nil
}

func overlap(a, b string) bool {
	// 检查两个代码块是否有重叠
	return false
}

// aiSemanticMerge 使用AI模型进行语义合并
func (e *SemanticMergeEngine) aiSemanticMerge(
	ctx context.Context,
	region ConflictRegion,
) (MergeResult, error) {
	// 构建prompt发送给代码理解模型
	prompt := fmt.Sprintf(`You are a code merge assistant.
Base code:
%s

Our changes:
%s

Their changes:
%s

Please produce the merged result.`, region.Base, region.Ours, region.Theirs)
	
	// 调用AI模型接口
	// response := callModel(ctx, e.modelEndpoint, prompt)
	
	return MergeResult{}, fmt.Errorf("not implemented")
}

// 自动回滚触发条件
func (e *SemanticMergeEngine) autoRollback(
	ctx context.Context,
	mergedContent string,
	unitTests []string,
) bool {
	// 运行单元测试检查合并结果
	for _, test := range unitTests {
		if !e.runTest(ctx, test, mergedContent) {
			// 回滚变更
			return true
		}
	}
	return false
}

# .github/workflows/ci.yml - 兼容Origin的GitHub Actions工作流
name: Agent CI Pipeline
on:
  push:
    branches: [ "main", "feature/*", "fix/*" ]
  pull_request:
    branches: [ "main" ]

env:
  PYTHON_VERSION: '3.12'
  NODE_VERSION: '20'

jobs:
  lint-and-typecheck:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      
      - name: Setup Python
        uses: actions/setup-python@v5
        with:
          python-version: ${{ env.PYTHON_VERSION }}
          
      - name: Install dependencies
        run: |
          python -m pip install --upgrade pip
          pip install ruff mypy pytest
          
      - name: Lint check
        run: ruff check src/ --output-format=github
        
      - name: Type check
        run: mypy src/ --strict
        
  test:
    needs: lint-and-typecheck
    runs-on: ubuntu-latest
    strategy:
      matrix:
        python-version: ['3.11', '3.12']
        
    steps:
      - uses: actions/checkout@v4
      
      - name: Setup Python ${{ matrix.python-version }}
        uses: actions/setup-python@v5
        with:
          python-version: ${{ matrix.python-version }}
          
      - name: Install dependencies
        run: pip install -r requirements.txt
        
      - name: Run unit tests
        run: pytest tests/unit/ --cov=src/ --cov-report=xml -x
        
      - name: Run integration tests
        run: pytest tests/integration/ --timeout=300
        
      - name: Upload coverage
        uses: codecov/codecov-action@v4
        with:
          file: ./coverage.xml
          fail_ci_if_error: false

  agent-validation:
    needs: test
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      
      - name: Validate AI Agent compatibility
        run: |
          # 检查代码是否可以被Agent正确理解和修改
          python -c "
          import ast
          import os
          
          errors = []
          for root, dirs, files in os.walk('src/'):
              for f in files:
                  if f.endswith('.py'):
                      path = os.path.join(root, f)
                      try:
                          with open(path) as fh:
                              ast.parse(fh.read())
                      except SyntaxError as e:
                          errors.append(f'{path}: {e}')
          
          if errors:
              print('Syntax errors found:')
              for e in errors:
                  print(f'  - {e}')
              exit(1)
          print('All files validated for agent compatibility')
          "

  security-scan:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      
      - name: Security scan
        run: |
          pip install bandit safety
          bandit -r src/ -f json -o security-report.json || true
          safety check -r requirements.txt --full-report || true

# .buildkite/pipeline.yml
steps:
  - label: "🔍 Lint & Type Check"
    command: |
      pip install ruff mypy
      ruff check src/ --output-format=github
      mypy src/ --strict
    agents:
      queue: "default"
    timeout_in_minutes: 10

  - label: "🧪 Unit Tests (Parallel)"
    command: |
      pip install -r requirements.txt
      pytest tests/unit/ -n auto --dist loadgroup \
        --junitxml=test-results.xml
    parallelism: 4
    artifact_paths:
      - "test-results.xml"
    timeout_in_minutes: 15
    env:
      PYTHONWARNINGS: "ignore"

  - label: "🔗 Integration Tests"
    command: |
      pip install -r requirements.txt
      docker compose -f docker-compose.test.yml up -d
      sleep 10
      pytest tests/integration/ --timeout=300
    depends_on: "🧪 Unit Tests (Parallel)"
    timeout_in_minutes: 30
    plugins:
      - docker-compose#v4.0.0:
          run: app

  - label: "🚀 Deploy Preview (Vercel)"
    command: |
      npx vercel --token $VERCEL_TOKEN \
        --scope $VERCEL_SCOPE \
        --confirm
    depends_on: "🔗 Integration Tests"
    branches: "feature/*"
    timeout_in_minutes: 20
    env:
      VERCEL_ORG_ID: "team_xxxx"
      VERCEL_PROJECT_ID: "prj_xxxx"

  - block: "👀 Human Review Gate"
    prompt: "Review all changes before production deployment"
    branches: "main"
    depends_on:
      - "🔍 Lint & Type Check"
      - "🧪 Unit Tests (Parallel)"
      - "🔗 Integration Tests"

  - label: "🚀 Deploy Production"
    command: |
      echo "Deploying to production..."
      ./scripts/deploy.sh
    depends_on: "👀 Human Review Gate"
    branches: "main"
    timeout_in_minutes: 30

#!/usr/bin/env python3
"""
示例：Agent自动检测变更并触发部署流水线
"""
import os
import json
import requests
from datetime import datetime
from typing import Optional

class AutoDeployAgent:
    """自动部署Agent"""
    
    def __init__(self, origin_token: str, buildkite_token: str):
        self.origin_token = origin_token
        self.buildkite_token = buildkite_token
        self.origin_api = "https://api.cursor.com/v1"
        self.buildkite_api = "https://api.buildkite.com/v2"
    
    def check_merged_prs(self, codebase: str, repo: str) -> list:
        """检查最近合并的PR"""
        url = f"{self.origin_api}/codebase/{codebase}/repos/{repo}/pulls"
        params = {"state": "merged", "per_page": 10}
        
        response = requests.get(
            url, params=params,
            headers={"Authorization": f"Bearer {self.origin_token}"}
        )
        response.raise_for_status()
        return response.json()
    
    def analyze_change_impact(self, pr: dict) -> dict:
        """分析变更影响范围"""
        changes = pr.get("changed_files", [])
        
        impact = {
            "has_database_migration": False,
            "has_api_changes": False,
            "has_config_changes": False,
            "affected_services": set(),
            "risk_level": "low"
        }
        
        for change in changes:
            filename = change["filename"]
            
            if "migration" in filename or "alembic" in filename:
                impact["has_database_migration"] = True
                impact["affected_services"].add("database")
            
            if "/api/" in filename or "routes" in filename:
                impact["has_api_changes"] = True
                impact["affected_services"].add("api")
            
            if filename.endswith((".yaml", ".yml", ".env")):
                impact["has_config_changes"] = True
                impact["affected_services"].add("config")
            
            # 检测服务模块
            parts = filename.split("/")
            if len(parts) >= 2 and parts[0] == "services":
                impact["affected_services"].add(parts[1])
        
        # 风险评估
        if impact["has_database_migration"]:
            impact["risk_level"] = "high"
        elif impact["has_api_changes"]:
            impact["risk_level"] = "medium"
        
        impact["affected_services"] = list(impact["affected_services"])
        return impact
    
    def trigger_buildkite_pipeline(
        self,
        pipeline_slug: str,
        branch: str,
        commit: str,
        message: str,
        env: Optional[dict] = None
    ) -> dict:
        """触发Buildkite构建"""
        org_slug = "acme-corp"
        url = f"{self.buildkite_api}/organizations/{org_slug}/pipelines/{pipeline_slug}/builds"
        
        payload = {
            "commit": commit,
            "branch": branch,
            "message": message,
            "env": env or {}
        }
        
        response = requests.post(
            url, json=payload,
            headers={"Authorization": f"Bearer {self.buildkite_token}"}
        )
        response.raise_for_status()
        return response.json()
    
    def run(self, codebase: str, repo: str):
        """主执行流程"""
        print(f"🚀 AutoDeploy Agent started at {datetime.now()}")
        print(f"📦 Checking {codebase}/{repo} for merged PRs...")
        
        merged_prs = self.check_merged_prs(codebase, repo)
        
        for pr in merged_prs:
            print(f"\n{'='*60}")
            print(f"📋 PR #{pr['number']}: {pr['title']}")
            print(f"   Branch: {pr['head']['ref']} → {pr['base']['ref']}")
            print(f"   Merged by: {pr['merged_by']['login']}")
            
            # 分析变更影响
            impact = self.analyze_change_impact(pr)
            print(f"🔍 Impact Analysis:")
            print(f"   - Risk Level: {impact['risk_level']}")
            print(f"   - Database Migration: {impact['has_database_migration']}")
            print(f"   - API Changes: {impact['has_api_changes']}")
            print(f"   - Affected Services: {', '.join(impact['affected_services'])}")
            
            # 根据风险级别决定部署策略
            if impact["risk_level"] == "high":
                print("⚠️  High risk change - triggering staged deployment")
                env = {
                    "DEPLOY_STRATEGY": "blue-green",
                    "CANARY_PERCENTAGE": "10",
                    "AUTO_ROLLBACK_ENABLED": "true",
                    "REQUIRE_MANUAL_APPROVAL": "true"
                }
            elif impact["risk_level"] == "medium":
                print("⚡ Medium risk change - triggering standard deployment")
                env = {
                    "DEPLOY_STRATEGY": "rolling",
                    "AUTO_ROLLBACK_ENABLED": "true"
                }
            else:
                print("✅ Low risk change - triggering fast deployment")
                env = {
                    "DEPLOY_STRATEGY": "fast",
                    "SKIP_HEAVY_TESTS": "true"
                }
            
            # 触发CI/CD
            build = self.trigger_buildkite_pipeline(
                pipeline_slug="deploy-pipeline",
                branch=pr["head"]["ref"],
                commit=pr["merge_commit_sha"],
                message=f"Auto-deploy: PR #{pr['number']} - {pr['title']}",
                env=env
            )
            
            print(f"🔨 Build triggered: {build['web_url']}")
        
        print(f"\n✅ AutoDeploy Agent completed at {datetime.now()}")


if __name__ == "__main__":
    agent = AutoDeployAgent(
        origin_token=os.environ["ORIGIN_TOKEN"],
        buildkite_token=os.environ["BUILDKITE_TOKEN"]
    )
    agent.run("acme-corp", "backend-api")