OpenAI ChatGPT Work + 10M Weekly Active Agents: The Milestone from Tech Validation to Mass Commercialization

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


class AgentRole(Enum):
    GENERAL = "general"
    CODER = "coder"
    ANALYST = "analyst"
    CUSTOMER = "customer"


@dataclass
class AgentTask:
    task_id: str
    role: AgentRole
    prompt: str
    context: Dict[str, Any] = field(default_factory=dict)
    tools: List[str] = field(default_factory=list)
    max_steps: int = 20
    created_at: float = field(default_factory=time.time)
    status: str = "pending"
    result: Optional[Dict] = None
    error: Optional[str] = None


class EnterpriseAgentRuntime:
    """
    Enterprise-grade Agent runtime
    Supports multi-Agent collaboration, permission management, audit logging
    """
    def __init__(self, org_id: str):
        self.org_id = org_id
        self.agents: Dict[AgentRole, 'BaseAgent'] = {}
        self.task_queue: asyncio.Queue = asyncio.Queue()
        self.active_tasks: Dict[str, AgentTask] = {}
        self.completed_tasks: List[AgentTask] = []
        self.audit_log: List[Dict] = []
        self.permission_manager = PermissionManager(org_id)
        self.context_store = ContextStore()
        self.stats = AgentStats()

    def register_agent(self, role: AgentRole, agent: 'BaseAgent'):
        self.agents[role] = agent
        agent.runtime = self

    async def submit_task(self, task: AgentTask) -> str:
        if not self.permission_manager.check_permission(task.role, task.prompt):
            task.status = "failed"
            task.error = "Insufficient permissions"
            return task.task_id

        self._log_audit("task_submitted", {
            "task_id": task.task_id,
            "role": task.role.value,
            "prompt_preview": task.prompt[:100]
        })

        await self.task_queue.put(task)
        self.active_tasks[task.task_id] = task
        self.stats.total_tasks += 1
        return task.task_id

    async def execute_loop(self):
        while True:
            task = await self.task_queue.get()
            agent = self.agents.get(task.role)
            if not agent:
                task.status = "failed"
                task.error = f"No agent found for role {task.role}"
                continue

            try:
                task.status = "running"
                start_time = time.time()
                result = await agent.execute(task)
                task.status = "completed"
                task.result = result
                elapsed = time.time() - start_time
                self.stats.total_execution_time += elapsed
                self.stats.completed_tasks += 1
                self._log_audit("task_completed", {
                    "task_id": task.task_id,
                    "elapsed_seconds": elapsed
                })
            except Exception as e:
                task.status = "failed"
                task.error = str(e)
                self.stats.failed_tasks += 1

            self.completed_tasks.append(task)
            del self.active_tasks[task.task_id]
            self.task_queue.task_done()

    def _log_audit(self, action: str, details: Dict):
        self.audit_log.append({
            "timestamp": time.time(),
            "org_id": self.org_id,
            "action": action,
            "details": details
        })

    def get_org_stats(self) -> Dict:
        return {
            "org_id": self.org_id,
            "total_tasks": self.stats.total_tasks,
            "completed": self.stats.completed_tasks,
            "failed": self.stats.failed_tasks,
            "avg_execution_time": (
                self.stats.total_execution_time / self.stats.completed_tasks
                if self.stats.completed_tasks > 0 else 0
            ),
            "active_agents": len(self.agents)
        }


@dataclass
class AgentStats:
    total_tasks: int = 0
    completed_tasks: int = 0
    failed_tasks: int = 0
    total_execution_time: float = 0.0


class PermissionManager:
    def __init__(self, org_id: str):
        self.org_id = org_id
        self.role_permissions: Dict[str, set] = {}

    def set_role_permission(self, role: str, tool: str):
        if role not in self.role_permissions:
            self.role_permissions[role] = set()
        self.role_permissions[role].add(tool)

    def check_permission(self, agent_role: AgentRole, prompt: str) -> bool:
        role_name = agent_role.value
        if role_name not in self.role_permissions:
            return False

        sensitive_patterns = ["delete", "drop", "truncate", "rm -rf",
                            "shutdown", "reboot", "format"]
        prompt_lower = prompt.lower()
        for pattern in sensitive_patterns:
            if pattern in prompt_lower:
                if role_name in self.role_permissions:
                    permitted = any(pattern in perm.lower()
                                     for perm in self.role_permissions[role_name])
                    if not permitted:
                        return False
        return True


class ContextStore:
    def __init__(self):
        self.short_term: Dict[str, List[Dict]] = {}
        self.long_term: Dict[str, Dict] = {}

    def add_to_context(self, session_id: str, entry: Dict):
        if session_id not in self.short_term:
            self.short_term[session_id] = []
        self.short_term[session_id].append({**entry, "timestamp": time.time()})
        if len(self.short_term[session_id]) > 100:
            self.short_term[session_id] = self.short_term[session_id][-50:]


class BaseAgent:
    def __init__(self, role: AgentRole, model: str = "gpt-5.6-luna"):
        self.role = role
        self.model = model
        self.runtime: Optional[EnterpriseAgentRuntime] = None

    async def execute(self, task: AgentTask) -> Dict:
        raise NotImplementedError

    async def call_llm(self, prompt: str) -> str:
        await asyncio.sleep(0.1)
        return f"Response for: {prompt[:50]}..."


class GeneralAgent(BaseAgent):
    def __init__(self):
        super().__init__(AgentRole.GENERAL)

    async def execute(self, task: AgentTask) -> Dict:
        steps = []
        current_prompt = task.prompt
        for step in range(task.max_steps):
            plan = await self._decompose_task(current_prompt)
            steps.append({"step": step, "plan": plan})
            if plan.get("is_complete"):
                break
            current_prompt = f"Continue: {task.prompt}"
        return {"steps": steps, "final_output": steps[-1]["plan"].get("result", "") if steps else ""}

    async def _decompose_task(self, prompt: str) -> Dict:
        await asyncio.sleep(0.05)
        return {"action": "execute", "is_complete": True, "result": f"Completed: {prompt[:30]}..."}


class CodexAgent(BaseAgent):
    def __init__(self):
        super().__init__(AgentRole.CODER, "gpt-5.6-terra")

    async def execute(self, task: AgentTask) -> Dict:
        code = await self._generate_code(task.prompt)
        review = await self._review_code(code)
        tests = await self._generate_tests(code)
        return {"code": code, "review": review, "tests": tests}

    async def _generate_code(self, prompt: str) -> str:
        await asyncio.sleep(0.2)
        return f"# Generated code for: {prompt[:50]}"

    async def _review_code(self, code: str) -> Dict:
        await asyncio.sleep(0.1)
        return {"approved": True, "issues": [], "suggestions": ["Add error handling"]}

    async def _generate_tests(self, code: str) -> List[str]:
        await asyncio.sleep(0.1)
        return ["test_1: basic functionality", "test_2: edge cases"]


async def simulate_enterprise_deployment():
    runtime = EnterpriseAgentRuntime("org_sme_001")
    runtime.register_agent(AgentRole.GENERAL, GeneralAgent())
    runtime.register_agent(AgentRole.CODER, CodexAgent())
    runtime.permission_manager.set_role_permission("general", "search")
    runtime.permission_manager.set_role_permission("coder", "code_generate")

    execution_task = asyncio.create_task(runtime.execute_loop())

    tasks = [
        AgentTask(task_id="t1", role=AgentRole.GENERAL, prompt="Search Q3 financial data"),
        AgentTask(task_id="t2", role=AgentRole.CODER, prompt="Write Python data cleaning function"),
        AgentTask(task_id="t3", role=AgentRole.GENERAL, prompt="Analyze customer feedback"),
    ]

    for task in tasks:
        await runtime.submit_task(task)

    await runtime.task_queue.join()
    stats = runtime.get_org_stats()
    print("Enterprise Agent Deployment Results:")
    print(f"  Total tasks: {stats['total_tasks']}")
    print(f"  Completed: {stats['completed']}")
    print(f"  Failed: {stats['failed']}")
    print(f"  Avg execution: {stats['avg_execution_time']:.4f}s")
    print(f"  Audit log entries: {len(runtime.audit_log)}")

    execution_task.cancel()


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

package main

import (
	"fmt"
	"math"
	"sync"
	"sync/atomic"
	"time"
)

type AgentScaler struct {
	mu            sync.RWMutex
	agentPools    map[string]*AgentPool
	config        ScalerConfig
	activeUsers   int64
	totalRequests int64
}

type ScalerConfig struct {
	MinAgentsPerPool int
	MaxAgentsPerPool int
	ScaleUpThreshold float64
	ScaleDownPeriod  time.Duration
}

type AgentPool struct {
	ID           string
	Agents       []*AgentInstance
	Capacity     int
	CurrentLoad  float64
	LastScaledAt time.Time
}

type AgentInstance struct {
	ID          string
	Model       string
	Status      string
	CurrentLoad float64
	TotalTokens int64
	ErrorCount  int64
}

func NewAgentScaler(config ScalerConfig) *AgentScaler {
	return &AgentScaler{
		agentPools: make(map[string]*AgentPool),
		config:     config,
	}
}

func (s *AgentScaler) GetOrCreatePool(model string) *AgentPool {
	s.mu.Lock()
	defer s.mu.Unlock()

	if pool, exists := s.agentPools[model]; exists {
		return pool
	}

	pool := &AgentPool{
		ID:     fmt.Sprintf("pool_%s_%d", model, time.Now().Unix()),
		Agents: make([]*AgentInstance, 0),
	}
	for i := 0; i < s.config.MinAgentsPerPool; i++ {
		pool.Agents = append(pool.Agents, &AgentInstance{
			ID:     fmt.Sprintf("%s_inst_%d", model, i),
			Model:  model,
			Status: "idle",
		})
	}
	s.agentPools[model] = pool
	return pool
}

func (s *AgentScaler) AutoScale() {
	s.mu.Lock()
	defer s.mu.Unlock()

	for model, pool := range s.agentPools {
		avgLoad := 0.0
		activeCount := 0
		for _, agent := range pool.Agents {
			if agent.Status != "offline" {
				avgLoad += agent.CurrentLoad
				activeCount++
			}
		}
		if activeCount > 0 {
			avgLoad /= float64(activeCount)
		}

		if avgLoad > s.config.ScaleUpThreshold && len(pool.Agents) < s.config.MaxAgentsPerPool {
			pool.Agents = append(pool.Agents, &AgentInstance{
				ID:     fmt.Sprintf("%s_inst_%d", model, len(pool.Agents)),
				Model:  model,
				Status: "active",
			})
			pool.LastScaledAt = time.Now()
			fmt.Printf("[ScaleUp] %s: load=%.2f, agents=%d\n", model, avgLoad, len(pool.Agents))
		}

		if avgLoad < s.config.ScaleUpThreshold*0.3 &&
			len(pool.Agents) > s.config.MinAgentsPerPool {
			pool.Agents = pool.Agents[:len(pool.Agents)-1]
			fmt.Printf("[ScaleDown] %s: agents=%d\n", model, len(pool.Agents))
		}
	}
}

func main() {
	config := ScalerConfig{
		MinAgentsPerPool: 10,
		MaxAgentsPerPool: 100,
		ScaleUpThreshold: 0.7,
		ScaleDownPeriod:  5 * time.Minute,
	}

	scaler := NewAgentScaler(config)
	pool := scaler.GetOrCreatePool("gpt-5.6-luna")

	for i := 0; i < 1000; i++ {
		if i%100 == 0 {
			scaler.AutoScale()
		}
		atomic.AddInt64(&pool.Agents[i%len(pool.Agents)].TotalTokens, 1)
	}

	fmt.Printf("\nFinal state:\n")
	fmt.Printf("  Pool: %s\n", pool.ID)
	fmt.Printf("  Agents: %d\n", len(pool.Agents))
}