Hosted Agents

Hosted Agent Infrastructure

Hosted agents run in remote sandboxed environments rather than on local machines. When designed well, they provide unlimited concurrency, consistent execution environments, and multiplayer collaboration. The critical insight is that session speed should be limited only by model provider time-to-first-token, with all infrastructure setup completed before the user starts their session.

When to Activate

Activate this skill when:

• Building background coding agents that run independently of user devices
• Designing sandboxed execution environments for agent workloads
• Implementing multiplayer agent sessions with shared state
• Creating multi-client agent interfaces (Slack, Web, Chrome extensions)
• Scaling agent infrastructure beyond local machine constraints
• Building systems where agents spawn sub-agents for parallel work

Core Concepts

Move agent execution to remote sandboxed environments to eliminate the fundamental limits of local execution: resource contention, environment inconsistency, and single-user constraints. Remote sandboxes unlock unlimited concurrency, reproducible environments, and collaborative workflows because each session gets its own isolated compute with a known-good environment image.

Design the architecture in three layers because each layer scales independently. Build sandbox infrastructure for isolated execution, an API layer for state management and client coordination, and client interfaces for user interaction across platforms. Keep these layers cleanly separated so sandbox changes do not ripple into clients.

Detailed Topics

Sandbox Infrastructure

The Core Challenge Eliminate sandbox spin-up latency because users perceive anything over a few seconds as broken. Development environments require cloning repositories, installing dependencies, and running build steps -- do all of this before the user ever submits a prompt.

Image Registry Pattern Pre-build environment images on a regular cadence (every 30 minutes works well) because this makes synchronization with the latest code a fast delta rather than a full clone. Include in each image:

• Cloned repository at a known commit
• All runtime dependencies installed
• Initial setup and build commands completed
• Cached files from running app and test suite once

When starting a session, spin up a sandbox from the most recent image. The repository is at most 30 minutes out of date, making the remaining git sync fast.

Snapshot and Restore Take filesystem snapshots at key points to enable instant restoration for follow-up prompts without re-running setup:

• After initial image build (base snapshot)
• When agent finishes making changes (session snapshot)
• Before sandbox exit for potential follow-up

Git Configuration for Background Agents Configure git identity explicitly in every sandbox because background agents are not tied to a specific user during image builds:

• Generate GitHub app installation tokens for repository access during clone
• Set git config user.name and user.email when committing and pushing changes
• Use the prompting user's identity for commits, not the app identity

Warm Pool Strategy Maintain a pool of pre-warmed sandboxes for high-volume repositories because cold starts are the primary source of user frustration:

• Keep sandboxes ready before users start sessions
• Expire and recreate pool entries as new image builds complete
• Start warming a sandbox as soon as a user begins typing (predictive warm-up)

Agent Framework Selection

Server-First Architecture Structure the agent framework as a server first, with TUI and desktop apps as thin clients, because this prevents duplicating agent logic across surfaces:

• Multiple custom clients share one agent backend
• Consistent behavior across all interaction surfaces
• Plugin systems extend functionality without client changes
• Event-driven architectures deliver real-time updates to any connected client

Code as Source of Truth Select frameworks where the agent can read its own source code to understand behavior. Prioritize this because having code as source of truth prevents the agent from hallucinating about its own capabilities -- an underrated failure mode in AI development.

Plugin System Requirements Require a plugin system that supports runtime interception because this enables safety controls and observability without modifying core agent logic:

• Listen to tool execution events (e.g., tool.execute.before)
• Block or modify tool calls conditionally
• Inject context or state at runtime

Speed Optimizations

Predictive Warm-Up Start warming the sandbox as soon as a user begins typing their prompt, not when they submit it, because the typing interval (5-30 seconds) is enough to complete most setup:

• Clone latest changes in parallel with user typing
• Run initial setup before user hits enter
• For fast spin-up, sandbox can be ready before user finishes typing

Parallel File Reading Allow the agent to start reading files immediately even if sync from latest base branch is not complete, because in large repositories incoming prompts rarely touch recently-changed files:

• Agent can research immediately without waiting for git sync
• Block file edits (not reads) until synchronization completes
• This separation is safe because read-time data staleness of 30 minutes rarely matters for research

Maximize Build-Time Work Move everything possible to the image build step because build-time duration is invisible to users:

• Full dependency installation
• Database schema setup
• Initial app and test suite runs (populates caches)

Self-Spawning Agents

Agent-Spawned Sessions Build tools that allow agents to spawn new sessions because frontier models are capable of decomposing work and coordinating sub-tasks:

• Research tasks across different repositories
• Parallel subtask execution for large changes
• Multiple smaller PRs from one major task

Expose three primitives: start a new session with specified parameters, read status of any session (check-in capability), and continue main work while sub-sessions run in parallel.

Prompt Engineering for Self-Spawning Engineer prompts that guide when agents should spawn sub-sessions rather than doing work inline:

• Research tasks that require cross-repository exploration
• Breaking monolithic changes into smaller PRs
• Parallel exploration of different approaches

API Layer

Per-Session State Isolation Isolate state per session (SQLite per session works well) because cross-session interference is a subtle and hard-to-debug failure mode:

• Dedicated database per session
• No session can impact another's performance
• Architecture handles hundreds of concurrent sessions

Real-Time Streaming Stream all agent work in real-time because high-frequency feedback is critical for user trust:

• Token streaming from model providers
• Tool execution status updates
• File change notifications

Use WebSocket connections with hibernation APIs to reduce compute costs during idle periods while maintaining open connections.

Synchronization Across Clients Build a single state system that synchronizes across all clients (chat interfaces, Slack bots, Chrome extensions, web interfaces, VS Code instances) because users switch surfaces frequently and expect continuity. All changes sync to the session state, enabling seamless client switching.

Multiplayer Support

Why Multiplayer Matters Design for multiplayer from day one because it is nearly free to add with proper synchronization architecture, and it unlocks high-value workflows:

• Teaching non-engineers to use AI effectively
• Live QA sessions with multiple team members
• Real-time PR review with immediate changes
• Collaborative debugging sessions

Implementation Requirements Build the data model so sessions are not tied to single authors because multiplayer fails silently if authorship is hardcoded:

• Pass authorship info to each prompt
• Attribute code changes to the prompting user
• Share session links for instant collaboration

Authentication and Authorization

User-Based Commits Use GitHub authentication to open PRs on behalf of the user (not the app) because this preserves the audit trail and prevents users from approving their own AI-generated changes:

• Obtain user tokens for PR creation
• PRs appear as authored by the human, not the bot

Sandbox-to-API Flow Follow this sequence because it keeps sandbox permissions minimal while letting the API handle sensitive operations: 1. Sandbox pushes changes (updating git user config) 2. Sandbox sends event to API with branch name and session ID 3. API uses user's GitHub token to create PR 4. GitHub webhooks notify API of PR events

Client Implementations

Slack Integration Prioritize Slack as the first distribution channel for internal adoption because it creates a virality loop as team members see others using it:

• No syntax required, natural chat interface
• Build a classifier (fast model with repo descriptions) to determine which repository to work in
• Include hints for common repositories; allow "unknown" for ambiguous cases

Web Interface Build a web interface with these features because it serves as the primary power-user surface:

• Real-time streaming of agent work on desktop and mobile
• Hosted VS Code instance running inside sandbox
• Streamed desktop view for visual verification
• Before/after screenshots for PRs
• Statistics page: sessions resulting in merged PRs (primary metric), usage over time, live "humans prompting" count

Chrome Extension Build a Chrome extension for non-engineering users because DOM and React internals extraction gives higher precision than raw screenshots at lower token cost:

• Sidebar chat interface with screenshot tool
• Extract DOM/React internals instead of raw images
• Distribute via managed device policy (bypasses Chrome Web Store)

Practical Guidance

Follow-Up Message Handling

Choose between queueing and inserting follow-up messages sent during execution. Prefer queueing because it is simpler to manage and lets users send thoughts on next steps while the agent works. Build a mechanism to stop the agent mid-execution when needed, because without it users feel trapped.

Metrics That Matter

Track these metrics because they indicate real value rather than vanity usage:

• Sessions resulting in merged PRs (primary success metric)
• Time from session start to first model response
• PR approval rate and revision count
• Agent-written code percentage across repositories

Adoption Strategy

Drive internal adoption through visibility rather than mandates because forced usage breeds resentment:

• Work in public spaces (Slack channels) for visibility
• Let the product create virality loops
• Do not force usage over existing tools
• Build to people's needs, not hypothetical requirements

Guidelines

1. Pre-build environment images on regular cadence (30 minutes is a good default) 2. Start warming sandboxes when users begin typing, not when they submit 3. Allow file reads before git sync completes; block only writes 4. Structure agent framework as server-first with clients as thin wrappers 5. Isolate state per session to prevent cross-session interference 6. Attribute commits to the user who prompted, not the app 7. Track merged PRs as primary success metric 8. Build for multiplayer from the start; it is nearly free with proper sync architecture

Gotchas

1. Cold start latency: First sandbox spin-up takes 30-60s and users perceive this as broken. Use warm pools and predictive warm-up on keystroke to eliminate perceived wait time. 2. Image staleness: Infrequent image rebuilds mean agents run with outdated dependencies or code. Set a 30-minute rebuild cadence and monitor image age; alert if builds fail silently. 3. Sandbox cost runaway: Long-running agents without timeout or budget caps accumulate unexpected costs. Set hard timeout limits (default 4 hours) and per-session cost ceilings. 4. Auth token expiration mid-session: Long tasks fail when GitHub tokens expire partway through. Implement token refresh logic and check token validity before sensitive operations like PR creation. 5. Git config in sandboxes: Missing user.name or user.email causes commit failures in background agents. Always set git identity explicitly during sandbox configuration, never assume it carries over from the image. 6. State loss on sandbox recycle: Agents lose completed work if the sandbox is recycled or times out before results are extracted. Always snapshot before termination and extract artifacts (branches, PRs, files) before letting the sandbox die. 7. Oversubscribing warm pools: Maintaining too many warm sandboxes wastes money during low-traffic periods. Scale pool size based on traffic patterns and time-of-day; use autoscaling rather than fixed pool sizes. 8. Missing output extraction: Agents complete work inside the sandbox but results never get pulled out to the user. Build explicit extraction steps (push branch, create PR, return file contents) into the session teardown flow.

Integration

This skill builds on multi-agent-patterns for agent coordination and tool-design for agent-tool interfaces. It connects to:

• multi-agent-patterns - Self-spawning agents follow supervisor patterns
• tool-design - Building tools for agent spawning and status checking
• context-optimization - Managing context across distributed sessions
• filesystem-context - Using filesystem for session state and artifacts

References

Internal reference:

• Infrastructure Patterns - Read when: implementing sandbox lifecycle, image builds, or warm pool logic for the first time

Related skills in this collection:

• multi-agent-patterns - Read when: designing self-spawning or supervisor coordination patterns
• tool-design - Read when: building tools for agent session management or status checking
• context-optimization - Read when: context windows fill up across distributed agent sessions

External resources:

• Ramp - Read when: evaluating whether to build vs. buy background agent infrastructure
• Modal Sandboxes - Read when: choosing a cloud sandbox provider or comparing isolation models
• Cloudflare Durable Objects - Read when: designing per-session state management with WebSocket hibernation
• OpenCode - Read when: selecting a server-first agent framework or studying plugin architectures

---

Skill Metadata

Created: 2026-01-12 Last Updated: 2026-03-17 Author: Agent Skills for Context Engineering Contributors Version: 1.1.0

Infrastructure Patterns for Hosted Agents

This reference provides detailed implementation patterns for building hosted agent infrastructure. These patterns are derived from production systems at scale.

Sandbox Architecture

Modal Integration Pattern

Modal provides the sandbox infrastructure with near-instant startup and filesystem snapshots.

import modal

# Define the base image with all dependencies
image = modal.Image.debian_slim().pip_install([
    "opencode",
    "gitpython",
    "psycopg2-binary",
])

# Create the app
app = modal.App("coding-agent")

# Sandbox class with snapshot support
@app.cls(image=image, timeout=3600)
class AgentSandbox:
    def __init__(self, repo_url: str, snapshot_id: str = None):
        self.repo_url = repo_url
        self.snapshot_id = snapshot_id
    
    @modal.enter()
    def setup(self):
        if self.snapshot_id:
            # Restore from snapshot
            modal.Sandbox.restore(self.snapshot_id)
        else:
            # Fresh setup from image
            self._clone_and_setup()
    
    def _clone_and_setup(self):
        """Clone repo and run initial setup."""
        token = self._get_github_app_token()
        os.system(f"git clone https://x-access-token:{token}@github.com/{self.repo_url}")
        os.system("npm install")
        os.system("npm run build")
    
    @modal.method()
    def execute_prompt(self, prompt: str, user_identity: dict) -> dict:
        """Execute a prompt in the sandbox."""
        # Update git config for this user
        os.system(f'git config user.name "{user_identity["name"]}"')
        os.system(f'git config user.email "{user_identity["email"]}"')
        
        # Run the agent
        result = self.agent.run(prompt)
        
        return {
            "result": result,
            "snapshot_id": modal.Sandbox.snapshot()
        }

Image Build Pipeline

Build images on a schedule to keep them fresh:

import schedule
import time
from datetime import datetime

class ImageBuilder:
    def __init__(self, repositories: list[str]):
        self.repositories = repositories
        self.images = {}
    
    def build_all_images(self):
        """Build images for all repositories."""
        for repo in self.repositories:
            try:
                image = self._build_image(repo)
                self.images[repo] = {
                    "image": image,
                    "built_at": datetime.utcnow(),
                    "commit": self._get_latest_commit(repo)
                }
            except Exception as e:
                # Log but continue with other repos
                log.error(f"Failed to build image for {repo}: {e}")
    
    def _build_image(self, repo: str) -> str:
        """Build a single repository image."""
        sandbox = modal.Sandbox.create()
        
        # Clone with app token
        token = get_app_installation_token(repo)
        sandbox.exec(f"git clone https://x-access-token:{token}@github.com/{repo} /workspace")
        
        # Install dependencies
        sandbox.exec("cd /workspace && npm install")
        
        # Run build
        sandbox.exec("cd /workspace && npm run build")
        
        # Warm caches
        sandbox.exec("cd /workspace && npm run dev &")
        time.sleep(5)  # Let dev server start
        sandbox.exec("cd /workspace && npm test -- --run")
        
        # Create snapshot
        return sandbox.snapshot()
    
    def get_latest_image(self, repo: str) -> str:
        """Get the most recent image for a repository."""
        if repo not in self.images:
            raise ValueError(f"No image available for {repo}")
        return self.images[repo]["image"]

# Schedule builds every 30 minutes
builder = ImageBuilder(["org/frontend", "org/backend", "org/shared"])
schedule.every(30).minutes.do(builder.build_all_images)

Warm Pool Management

Maintain pre-warmed sandboxes for instant session starts:

from collections import defaultdict
from dataclasses import dataclass
from datetime import datetime, timedelta

@dataclass
class WarmSandbox:
    sandbox_id: str
    repo: str
    created_at: datetime
    image_version: str
    is_claimed: bool = False

class WarmPoolManager:
    def __init__(self, target_pool_size: int = 3):
        self.target_size = target_pool_size
        self.pools = defaultdict(list)  # repo -> [WarmSandbox]
        self.max_age = timedelta(minutes=25)  # Expire before next image build
    
    def get_warm_sandbox(self, repo: str) -> WarmSandbox | None:
        """Get a pre-warmed sandbox if available."""
        pool = self.pools[repo]
        
        for sandbox in pool:
            if not sandbox.is_claimed and self._is_valid(sandbox):
                sandbox.is_claimed = True
                return sandbox
        
        return None
    
    def _is_valid(self, sandbox: WarmSandbox) -> bool:
        """Check if sandbox is still valid."""
        age = datetime.utcnow() - sandbox.created_at
        current_image = self.image_builder.get_latest_image(sandbox.repo)
        
        return (
            age < self.max_age and
            sandbox.image_version == current_image
        )
    
    def maintain_pool(self, repo: str):
        """Ensure pool has target number of warm sandboxes."""
        # Remove expired sandboxes
        self.pools[repo] = [s for s in self.pools[repo] if self._is_valid(s)]
        
        # Add new sandboxes to reach target
        current_count = len([s for s in self.pools[repo] if not s.is_claimed])
        needed = self.target_size - current_count
        
        for _ in range(needed):
            sandbox = self._create_warm_sandbox(repo)
            self.pools[repo].append(sandbox)
    
    def _create_warm_sandbox(self, repo: str) -> WarmSandbox:
        """Create a new warm sandbox from latest image."""
        image = self.image_builder.get_latest_image(repo)
        sandbox_id = modal.Sandbox.create(image=image)
        
        # Sync to latest (runs in background)
        self._sync_to_latest(sandbox_id, repo)
        
        return WarmSandbox(
            sandbox_id=sandbox_id,
            repo=repo,
            created_at=datetime.utcnow(),
            image_version=image
        )

API Layer Patterns

Cloudflare Durable Objects for Session State

Each session gets its own Durable Object with isolated SQLite:

// Session Durable Object
export class SessionDO implements DurableObject {
  private storage: DurableObjectStorage;
  private sql: SqlStorage;
  private connections: Map<string, WebSocket> = new Map();

  constructor(ctx: DurableObjectState) {
    this.storage = ctx.storage;
    this.sql = ctx.storage.sql;
    this.initializeSchema();
  }

  private initializeSchema() {
    this.sql.exec(`
      CREATE TABLE IF NOT EXISTS messages (
        id INTEGER PRIMARY KEY,
        role TEXT NOT NULL,
        content TEXT NOT NULL,
        author_id TEXT,
        author_name TEXT,
        created_at TEXT DEFAULT CURRENT_TIMESTAMP
      );
      
      CREATE TABLE IF NOT EXISTS artifacts (
        id INTEGER PRIMARY KEY,
        type TEXT NOT NULL,
        path TEXT,
        content TEXT,
        created_at TEXT DEFAULT CURRENT_TIMESTAMP
      );
      
      CREATE TABLE IF NOT EXISTS events (
        id INTEGER PRIMARY KEY,
        type TEXT NOT NULL,
        data TEXT,
        created_at TEXT DEFAULT CURRENT_TIMESTAMP
      );
    `);
  }

  async fetch(request: Request): Promise<Response> {
    const url = new URL(request.url);

    if (request.headers.get("Upgrade") === "websocket") {
      return this.handleWebSocket(request);
    }

    switch (url.pathname) {
      case "/message":
        return this.handleMessage(request);
      case "/status":
        return this.getStatus();
      default:
        return new Response("Not found", { status: 404 });
    }
  }

  private handleWebSocket(request: Request): Response {
    const pair = new WebSocketPair();
    const [client, server] = Object.values(pair);

    const connectionId = crypto.randomUUID();
    this.connections.set(connectionId, server);

    server.accept();
    server.addEventListener("close", () => {
      this.connections.delete(connectionId);
    });

    return new Response(null, { status: 101, webSocket: client });
  }

  private broadcast(message: object) {
    const data = JSON.stringify(message);
    for (const ws of this.connections.values()) {
      ws.send(data);
    }
  }

  async handleMessage(request: Request): Promise<Response> {
    const { content, author } = await request.json();

    // Store message
    this.sql.exec(
      `INSERT INTO messages (role, content, author_id, author_name) VALUES (?, ?, ?, ?)`,
      ["user", content, author.id, author.name]
    );

    // Broadcast to all connected clients
    this.broadcast({
      type: "message",
      role: "user",
      content,
      author,
    });

    // Forward to sandbox for processing
    const result = await this.forwardToSandbox(content, author);

    return Response.json(result);
  }
}

Real-Time Event Streaming

Stream events from sandbox to all connected clients:

class EventStream {
  private sessionDO: DurableObjectStub;

  async streamFromSandbox(sandboxId: string, sessionId: string) {
    const sandbox = await modal.Sandbox.get(sandboxId);

    // Subscribe to sandbox events
    for await (const event of sandbox.events()) {
      // Forward to Durable Object for broadcast
      await this.sessionDO.fetch(
        new Request(`https://internal/event`, {
          method: "POST",
          body: JSON.stringify({
            type: event.type,
            data: event.data,
          }),
        })
      );
    }
  }
}

Client Integration Patterns

Slack Bot with Repository Classification

from slack_bolt import App
from slack_bolt.adapter.socket_mode import SocketModeHandler

app = App(token=os.environ["SLACK_BOT_TOKEN"])

# Repository descriptions for classification
REPO_DESCRIPTIONS = [
    {
        "name": "frontend-monorepo",
        "description": "React frontend application with dashboard, user portal, and admin interfaces",
        "hints": ["dashboard", "UI", "component", "page", "frontend"]
    },
    {
        "name": "backend-services",
        "description": "Node.js API services including auth, payments, and core business logic",
        "hints": ["API", "endpoint", "service", "backend", "database"]
    },
    {
        "name": "mobile-app",
        "description": "React Native mobile application for iOS and Android",
        "hints": ["mobile", "app", "iOS", "Android", "native"]
    }
]

async def classify_repository(message: str, channel: str, thread: list[str]) -> str:
    """Use fast model to classify which repo the message refers to."""
    prompt = f"""Classify which repository this message is about.

Message: {message}
Channel: #{channel}
Thread context: {' | '.join(thread[-3:])}

Repositories:
{json.dumps(REPO_DESCRIPTIONS, indent=2)}

Return ONLY the repository name, or "unknown" if unclear."""

    response = await openai.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": prompt}],
        max_tokens=50
    )
    
    return response.choices[0].message.content.strip()

@app.event("app_mention")
async def handle_mention(event, say, client):
    """Handle @mentions of the bot."""
    channel = event["channel"]
    message = event["text"]
    thread_ts = event.get("thread_ts", event["ts"])
    
    # Get thread context if in a thread
    thread_messages = []
    if "thread_ts" in event:
        result = await client.conversations_replies(
            channel=channel,
            ts=thread_ts
        )
        thread_messages = [m["text"] for m in result["messages"]]
    
    # Get channel info for context
    channel_info = await client.conversations_info(channel=channel)
    channel_name = channel_info["channel"]["name"]
    
    # Classify repository
    repo = await classify_repository(message, channel_name, thread_messages)
    
    if repo == "unknown":
        await say(
            text="I'm not sure which repository you're referring to. Could you specify?",
            thread_ts=thread_ts
        )
        return
    
    # Start session and process
    session = await start_session(repo, event["user"])
    
    await say(
        text=f":robot_face: Starting work in `{repo}`...",
        thread_ts=thread_ts
    )
    
    result = await session.process(message)
    
    # Post result with Block Kit formatting
    await say(
        blocks=format_result_blocks(result),
        thread_ts=thread_ts
    )

Chrome Extension DOM Extraction

Extract DOM structure instead of sending screenshots:

// content-script.ts
interface ElementInfo {
  tag: string;
  classes: string[];
  id?: string;
  text?: string;
  rect: DOMRect;
  reactComponent?: string;
}

function extractDOMInfo(element: Element): ElementInfo {
  // Get React component name if available
  let reactComponent: string | undefined;
  const fiberKey = Object.keys(element).find((key) =>
    key.startsWith("__reactFiber")
  );
  if (fiberKey) {
    const fiber = (element as any)[fiberKey];
    reactComponent = fiber?.type?.name || fiber?.type?.displayName;
  }

  return {
    tag: element.tagName.toLowerCase(),
    classes: Array.from(element.classList),
    id: element.id || undefined,
    text: element.textContent?.slice(0, 100),
    rect: element.getBoundingClientRect(),
    reactComponent,
  };
}

function extractSelectedArea(selection: DOMRect): ElementInfo[] {
  const elements: ElementInfo[] = [];

  // Find all elements within selection bounds
  document.querySelectorAll("*").forEach((el) => {
    const rect = el.getBoundingClientRect();
    if (
      rect.top >= selection.top &&
      rect.left >= selection.left &&
      rect.bottom <= selection.bottom &&
      rect.right <= selection.right
    ) {
      elements.push(extractDOMInfo(el));
    }
  });

  return elements;
}

// Message handler for sidebar
chrome.runtime.onMessage.addListener((request, sender, sendResponse) => {
  if (request.type === "EXTRACT_SELECTION") {
    const elements = extractSelectedArea(request.selection);
    sendResponse({ elements });
  }
});

Multiplayer Implementation

Authorship Tracking

Track which user made each change:

@dataclass
class PromptContext:
    content: str
    author: Author
    session_id: str
    timestamp: datetime

@dataclass
class Author:
    id: str
    name: str
    email: str
    github_token: str  # For PR creation

class MultiplayerSession:
    def __init__(self, session_id: str):
        self.session_id = session_id
        self.participants: dict[str, Author] = {}
        self.prompt_queue: list[PromptContext] = []
    
    def add_participant(self, author: Author):
        """Add a participant to the session."""
        self.participants[author.id] = author
        self.broadcast_event("participant_joined", author)
    
    async def process_prompt(self, prompt: PromptContext):
        """Process prompt with author attribution."""
        # Update git config for this author
        await self.sandbox.exec(
            f'git config user.name "{prompt.author.name}"'
        )
        await self.sandbox.exec(
            f'git config user.email "{prompt.author.email}"'
        )
        
        # Run agent
        result = await self.agent.run(prompt.content)
        
        # If changes were made, create PR with author's token
        if result.has_changes:
            await self.create_pr(
                branch=result.branch,
                author=prompt.author
            )
        
        return result
    
    async def create_pr(self, branch: str, author: Author):
        """Create PR using the author's GitHub token."""
        async with aiohttp.ClientSession() as session:
            headers = {
                "Authorization": f"Bearer {author.github_token}",
                "Accept": "application/vnd.github.v3+json"
            }
            
            await session.post(
                f"https://api.github.com/repos/{self.repo}/pulls",
                headers=headers,
                json={
                    "title": self.generate_pr_title(),
                    "body": self.generate_pr_body(),
                    "head": branch,
                    "base": "main"
                }
            )

Metrics and Monitoring

Key Metrics to Track

from dataclasses import dataclass
from datetime import datetime, timedelta

@dataclass
class SessionMetrics:
    session_id: str
    started_at: datetime
    first_token_at: datetime | None
    completed_at: datetime | None
    pr_created: bool
    pr_merged: bool
    prompts_count: int
    participants_count: int
    
    @property
    def time_to_first_token(self) -> timedelta | None:
        if self.first_token_at:
            return self.first_token_at - self.started_at
        return None

class MetricsAggregator:
    def get_adoption_metrics(self, period: timedelta) -> dict:
        """Get adoption metrics for a time period."""
        sessions = self.get_sessions_in_period(period)
        
        total_prs = sum(1 for s in sessions if s.pr_created)
        merged_prs = sum(1 for s in sessions if s.pr_merged)
        
        return {
            "total_sessions": len(sessions),
            "prs_created": total_prs,
            "prs_merged": merged_prs,
            "merge_rate": merged_prs / total_prs if total_prs > 0 else 0,
            "avg_time_to_first_token": self._avg_ttft(sessions),
            "unique_users": len(set(s.author_id for s in sessions)),
            "multiplayer_sessions": sum(
                1 for s in sessions if s.participants_count > 1
            )
        }
    
    def get_repository_metrics(self) -> dict[str, dict]:
        """Get metrics broken down by repository."""
        metrics = {}
        
        for repo in self.repositories:
            repo_sessions = self.get_sessions_for_repo(repo)
            total_prs = self.get_total_prs(repo)
            agent_prs = sum(1 for s in repo_sessions if s.pr_merged)
            
            metrics[repo] = {
                "agent_pr_percentage": agent_prs / total_prs * 100,
                "session_count": len(repo_sessions),
                "avg_prompts_per_session": sum(
                    s.prompts_count for s in repo_sessions
                ) / len(repo_sessions)
            }
        
        return metrics

Security Considerations

Sandbox Isolation

class SandboxSecurityConfig:
    """Security configuration for sandboxes."""
    
    # Network restrictions
    allowed_hosts = [
        "github.com",
        "api.github.com",
        "registry.npmjs.org",
        "pypi.org",
    ]
    
    # Resource limits
    max_memory_mb = 4096
    max_cpu_cores = 2
    max_disk_gb = 10
    max_runtime_hours = 4
    
    # Secrets handling
    secrets_to_inject = [
        "GITHUB_APP_TOKEN",
        "NPM_TOKEN",
    ]
    
    # Blocked operations
    blocked_commands = [
        "curl",  # Use fetch tools instead
        "wget",
        "ssh",
    ]

Token Handling

class TokenManager:
    """Manage tokens for GitHub operations."""
    
    def get_app_installation_token(self, repo: str) -> str:
        """Get short-lived token for repo access."""
        # Token expires in 1 hour
        return github_app.create_installation_token(
            installation_id=self.get_installation_id(repo),
            permissions={"contents": "write", "pull_requests": "write"}
        )
    
    def get_user_token(self, user_id: str) -> str:
        """Get user's OAuth token for PR creation."""
        # Stored encrypted, decrypted at runtime
        encrypted = self.storage.get(f"user_token:{user_id}")
        return self.decrypt(encrypted)

References

• Modal Documentation
• Cloudflare Durable Objects
• Cloudflare Agents SDK
• GitHub Apps Authentication
• Slack Bolt for Python
• Chrome Extension APIs