# Agent-Computer Interface (ACI)

Bridging agents with computer interfaces: desktop control, OS integration, and computer use.

---

## ACI Fundamentals

```
Agent-Computer Interface (ACI)
├── Desktop Control
│   ├── Screen Observation
│   ├── Input Injection (Mouse/Keyboard)
│   ├── Window Management
│   └── Accessibility APIs
├── OS Integration
│   ├── File System Operations
│   ├── Process Management
│   ├── Shell/Terminal Access
│   └── System APIs
├── Application Control
│   ├── Browser Automation
│   ├── Office/IDE Control
│   ├── API Interaction
│   └── Clipboard/OS Services
├── Security & Isolation
│   ├── Sandboxing
│   ├── Permission Model
│   ├── Audit Logging
│   └── Remote Access
└── Observability
    ├── Screen Recording
    ├── Action Logging
    ├── Performance Metrics
    └── Debugging Interface
```

---

## Desktop Control

### Screen Observation
```python
class ScreenObserver:
    def __init__(self, config: ObservationConfig = None):
        self.config = config or ObservationConfig()
        self.capture = ScreenCapture()
    
    async def capture(self, region: Region = None) -> ScreenCapture:
        """Capture screen with configurable options"""
        if region:
            return await self.capture.capture_region(region)
        return await self.capture.capture_full()
    
    async def capture_stream(self, fps: int = 30) -> AsyncIterator[ScreenCapture]:
        """Stream screen captures at specified FPS"""
        interval = 1.0 / fps
        while True:
            start = time.time()
            capture = await self.capture()
            yield capture
            elapsed = time.time() - start
            await asyncio.sleep(max(0, interval - elapsed))

class ScreenCapture:
    def __init__(self):
        # Platform-specific backends
        self.backends = {
            'linux': X11Capture(),
            'darwin': QuartzCapture(),
            'windows': WindowsCapture()
        }
        self.backend = self.backends[platform.system().lower()]
    
    async def capture_full(self) -> ScreenCapture:
        return await self.backend.capture_full()
    
    async def capture_region(self, region: Region) -> ScreenCapture:
        return await self.backend.capture_region(region)
    
    async def capture_window(self, window_id: str) -> ScreenCapture:
        return await self.backend.capture_window(window_id)
```

### Input Injection
```python
class InputInjector:
    def __init__(self):
        self.backends = {
            'linux': X11Input(),
            'darwin': QuartzInput(),
            'windows': WindowsInput()
        }
        self.injector = self.backends[platform.system().lower()]
    
    async def click(self, x: int, y: int, button: str = 'left', count: int = 1):
        await self.injector.click(x, y, button, count)
    
    async def type_text(self, text: str, interval: float = 0.01):
        await self.injector.type_text(text, interval)
    
    async def key_press(self, key: str, modifiers: list[str] = None):
        await self.injector.key_press(key, modifiers)
    
    async def key_combination(self, keys: list[str]):
        await self.injector.key_combination(keys)
    
    async def scroll(self, x: int, y: int, dx: int, dy: int):
        await self.injector.scroll(x, y, dx, dy)
    
    async def drag_drop(self, start: Point, end: Point):
        await self.injector.drag_drop(start, end)

class X11Input:
    def __init__(self):
        self.display = Xlib.display.Display()
    
    async def click(self, x: int, y: int, button: str, count: int):
        # X11 button events
        button_map = {'left': 1, 'middle': 2, 'right': 3, 'wheel_up': 4, 'wheel_down': 5}
        btn = button_map.get(button, 1)
        
        for _ in range(count):
            self.display.screen().root.warp_pointer(x, y)
            self.display.send_event(
                Xlib.X.ButtonPress,
                detail=btn,
                root=self.display.screen().root,
                time=Xlib.X.CurrentTime
            )
            self.display.send_event(
                Xlib.X.ButtonRelease,
                detail=btn,
                root=self.display.screen().root,
                time=Xlib.X.CurrentTime
            )
        self.display.flush()
    
    async def type_text(self, text: str, interval: float):
        for char in text:
            keysym = Xlib.XK.string_to_keysym(char)
            keycode = self.display.keysym_to_keycode(keysym)
            self.inject_key(keycode, True)
            self.inject_key(keycode, False)
            await asyncio.sleep(interval)
```

### Window Management
```python
class WindowManager:
    def __init__(self):
        self.backends = {
            'linux': X11WindowManager(),
            'darwin': QuartzWindowManager(),
            'windows': WindowsWindowManager()
        }
        self.wm = self.backends[platform.system().lower()]
    
    async def list_windows(self) -> list[WindowInfo]:
        return await self.wm.list_windows()
    
    async def get_active_window(self) -> WindowInfo:
        return await self.wm.get_active_window()
    
    async def focus_window(self, window_id: str):
        await self.wm.focus_window(window_id)
    
    async def get_window_geometry(self, window_id: str) -> Geometry:
        return await self.wm.get_geometry(window_id)
    
    async def set_window_state(self, window_id: str, state: WindowState):
        await self.wm.set_state(window_id, state)
    
    async def capture_window(self, window_id: str) -> ScreenCapture:
        return await self.wm.capture(window_id)

class X11WindowManager:
    def __init__(self):
        self.display = Xlib.display.Display()
    
    async def list_windows(self) -> list[WindowInfo]:
        root = self.display.screen().root
        windows = []
        
        def traverse(window):
            try:
                attrs = window.get_attributes()
                if attrs.map_state == Xlib.X.IsViewable:
                    geom = window.get_geometry()
                    name = window.get_wm_name() or ""
                    windows.append(WindowInfo(
                        id=str(window.id),
                        title=name,
                        geometry=Geometry(geom.x, geom.y, geom.width, geom.height),
                        pid=self._get_pid(window)
                    ))
            except:
                pass
            for child in window.query_tree().children:
                traverse(child)
        
        traverse(root)
        return windows
```

---

## OS Integration

### File System Operations
```python
class FileSystemAgent:
    def __init__(self, sandbox: Sandbox = None):
        self.sandbox = sandbox or LocalSandbox()
    
    async def read_file(self, path: str, encoding: str = 'utf-8') -> str:
        path = self._resolve_path(path)
        async with aiofiles.open(path, 'r', encoding=encoding) as f:
            return await f.read()
    
    async def write_file(self, path: str, content: str, encoding: str = 'utf-8'):
        path = self._resolve_path(path)
        async with aiofiles.open(path, 'w', encoding=encoding) as f:
            await f.write(content)
    
    async def list_directory(self, path: str, recursive: bool = False) -> list[FileInfo]:
        path = self._resolve_path(path)
        if recursive:
            files = []
            for root, dirs, files in os.walk(path):
                for f in files:
                    path = os.path.join(root, f)
                    stat = os.stat(path)
                    files.append(FileInfo(path=path, size=stat.st_size, modified=stat.st_mtime))
            return files
        else:
            entries = await aiofiles.os.scandir(path)
            return [FileInfo.from_dirent(e) for e in entries]
    
    async def search_files(self, pattern: str, path: str = '.') -> list[str]:
        path = self._resolve_path(path)
        return await asyncio.to_thread(glob.glob, os.path.join(path, pattern), recursive=True)
    
    async def grep(self, pattern: str, path: str = '.', file_pattern: str = '*') -> list[GrepResult]:
        path = self._resolve_path(path)
        cmd = ['rg', '--json', pattern, path, '--glob', file_pattern]
        result = await asyncio.create_subprocess_exec(
            *cmd, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE
        )
        stdout, _ = await cmd.communicate()
        return [GrepResult.from_json(line) for line in stdout.decode().splitlines() if line]

class LocalSandbox:
    def __init__(self, root: str = '/tmp/agent_sandbox'):
        self.root = Path(root)
        self.root.mkdir(parents=True, exist_ok=True)
    
    def _resolve_path(self, path: str) -> Path:
        path = Path(path).expanduser()
        if not path.is_absolute():
            path = self.root / path
        # Ensure path is within sandbox
        try:
            path.resolve().relative_to(self.root.resolve())
        except ValueError:
            raise SecurityError(f"Path {path} escapes sandbox")
        return path
```

### Process Management
```python
class ProcessManager:
    def __init__(self):
        self.processes: dict[str, ProcessInfo] = {}
    
    async def spawn(self, command: str, args: list[str] = None, 
                    env: dict = None, cwd: str = None) -> ProcessHandle:
        proc = await asyncio.create_subprocess_exec(
            command, *(args or []),
            env={**os.environ, **(env or {})},
            cwd=cwd,
            stdout=asyncio.subprocess.PIPE,
            stderr=asyncio.subprocess.PIPE
        )
        
        handle = ProcessHandle(
            pid=proc.pid,
            command=command,
            args=args,
            process=proc,
            stdout=asyncio.Queue(),
            stderr=asyncio.Queue()
        )
        
        # Start output readers
        asyncio.create_task(self._read_stream(proc.stdout, handle.stdout))
        asyncio.create_task(self._read_stream(proc.stderr, handle.stderr))
        
        self.processes[str(proc.pid)] = ProcessInfo(handle=handle, start_time=time.time())
        return handle
    
    async def execute(self, command: str, args: list[str] = None, 
                      timeout: float = 30.0) -> ProcessResult:
        handle = await self.spawn(command, args)
        try:
            stdout, stderr = await asyncio.wait_for(
                handle.wait(), timeout=timeout
            )
            return ProcessResult(
                exit_code=handle.process.returncode,
                stdout=stdout,
                stderr=stderr
            )
        except asyncio.TimeoutError:
            handle.process.kill()
            raise TimeoutError(f"Command timed out after {timeout}s")
    
    async def kill(self, pid: int):
        if pid in self.processes:
            self.processes[pid].handle.process.terminate()
            await self.processes[pid].handle.process.wait()
            del self.processes[pid]
```

### Shell/Terminal Access
```python
class TerminalAgent:
    def __init__(self):
        self.sessions: dict[str, TerminalSession] = {}
    
    async def create_session(self, session_id: str, shell: str = '/bin/bash') -> TerminalSession:
        # Use pty for interactive terminal
        master, slave = pty.openpty()
        
        process = await asyncio.create_subprocess_exec(
            shell,
            stdin=slave,
            stdout=slave,
            stderr=slave,
            preexec_fn=os.setsid
        )
        
        session = TerminalSession(
            id=session_id,
            master_fd=master,
            slave_fd=slave,
            process=process,
            output_buffer=""
        )
        
        # Start output reader
        asyncio.create_task(self._read_output(session))
        
        self.sessions[session_id] = session
        return session
    
    async def send_input(self, session_id: str, data: str):
        session = self.sessions[session_id]
        os.write(session.master_fd, data.encode())
    
    async def resize(self, session_id: str, rows: int, cols: int):
        session = self.sessions[session_id]
        fcntl.ioctl(session.master_fd, termios.TIOCSWINSZ, 
                    struct.pack('HHHH', rows, cols, 0, 0))
    
    async def close_session(self, session_id: str):
        session = self.sessions.pop(session_id)
        os.close(session.master_fd)
        session.process.terminate()
        await session.process.wait()

class WebTerminalAgent:
    """WebSocket-based terminal for web-based agents"""
    def __init__(self):
        self.terminal = None
    
    async def start(self, websocket: WebSocket):
        self.terminal = await TerminalSession.create_websocket(websocket)
    
    async def handle_message(self, message: str):
        if message['type'] == 'input':
            await self.terminal.write(message['data'])
        elif message['type'] == 'resize':
            await self.terminal.resize(message['rows'], message['cols'])
```

---

## Application Control

### Browser Automation
```python
class BrowserAgent:
    def __init__(self, headless: bool = True):
        self.playwright = None
        self.browser = None
        self.headless = headless
    
    async def start(self):
        self.playwright = await async_playwright().start()
        self.browser = await self.playwright.chromium.launch(
            headless=self.headless,
            args=['--no-sandbox', '--disable-dev-shm-usage']
        )
    
    async def new_page(self, url: str = None) -> Page:
        context = await self.browser.new_context(
            viewport={'width': 1920, 'height': 1080},
            user_agent='Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36'
        )
        page = await context.new_page()
        if url:
            await page.goto(url, wait_until='networkidle')
        return page
    
    async def close(self):
        if self.browser:
            await self.browser.close()
        if self.playwright:
            await self.playwright.stop()

class Page:
    async def click(self, selector: str):
        await self.page.click(selector)
    
    async def fill(self, selector: str, value: str):
        await self.page.fill(selector, value)
    
    async def type(self, selector: str, text: str, delay: int = 50):
        await self.page.type(selector, text, delay=delay)
    
    async def wait_for_selector(self, selector: str, timeout: int = 30000):
        await self.page.wait_for_selector(selector, timeout=timeout)
    
    async def evaluate(self, script: str) -> Any:
        return await self.page.evaluate(script)
    
    async def screenshot(self, path: str = None, full_page: bool = True) -> bytes:
        return await self.page.screenshot(path=path, full_page=full_page)
    
    async def pdf(self, path: str = None) -> bytes:
        return await self.page.pdf(path=path)
    
    async def wait_for_navigation(self, timeout: int = 30000):
        await self.page.wait_for_load_state('networkidle', timeout=timeout)
```

### Office/IDE Control
```python
class VSCodeAgent:
    def __init__(self):
        self.vscode = None
    
    async def connect(self, port: int = 9222):
        """Connect to VS Code via Chrome DevTools Protocol"""
        import websockets
        self.ws = await websockets.connect(f'ws://localhost:{port}')
        await self._initialize_cdp()
    
    async def open_file(self, path: str):
        await self.send_cdp('Runtime.evaluate', {
            'expression': f'vscode.workspace.openTextDocument("{path}").then(doc => vscode.window.showTextDocument(doc))'
        })
    
    async def get_diagnostics(self, file_path: str) -> list[Diagnostic]:
        result = await self.send_cdp('Runtime.evaluate', {
            'expression': f'vscode.languages.getDiagnostics(vscode.Uri.file("{file_path}"))'
        })
        return result['result']['value']
    
    async def apply_code_action(self, file_path: str, action: CodeAction):
        await self.send_cdp('Runtime.evaluate', {
            'expression': f'vscode.commands.executeCommand("editor.action.codeAction", "{action}")'
        })
    
    async def run_task(self, task_name: str):
        await self.send_cdp('Runtime.evaluate', {
            'expression': f'vscode.tasks.executeTask("{task_name}")'
        })

class IntelliJAgent:
    """Control IntelliJ IDEA via HTTP API"""
    def __init__(self, host: str = 'localhost', port: int = 63342):
        self.base_url = f'http://{host}:{port}'
        self.session = aiohttp.ClientSession()
    
    async def open_file(self, path: str):
        await self.session.post(f'{self.base_url}/api/file/open', json={'path': path})
    
    async def run_inspection(self, scope: str = 'project') -> list[InspectionResult]:
        async with self.session.post(f'{self.base_url}/api/inspect', json={'scope': scope}) as resp:
            return await resp.json()
```

---

## Security & Isolation

### Sandboxing
```python
class SandboxManager:
    def __init__(self):
        self.sandboxes: dict[str, Sandbox] = {}
    
    def create_sandbox(self, sandbox_id: str, config: SandboxConfig) -> Sandbox:
        if config.type == 'docker':
            sandbox = DockerSandbox(sandbox_id, config)
        elif config.type == 'gvisor':
            sandbox = GVisorSandbox(sandbox_id, config)
        elif config.type == 'firecracker':
            sandbox = FirecrackerSandbox(sandbox_id, config)
        elif config.type == 'wasm':
            sandbox = WasmSandbox(sandbox_id, config)
        else:
            sandbox = LocalSandbox(sandbox_id, config)
        
        self.sandboxes[sandbox_id] = sandbox
        return sandbox
    
    async def execute(self, sandbox_id: str, code: str, 
                      language: str = 'python') -> ExecutionResult:
        sandbox = self.sandboxes.get(sandbox_id)
        if not sandbox:
            raise ValueError(f"Sandbox {sandbox_id} not found")
        return await sandbox.execute(code, language)

class DockerSandbox:
    def __init__(self, sandbox_id: str, config: SandboxConfig):
        self.sandbox_id = sandbox_id
        self.config = config
        self.container = None
    
    async def start(self):
        self.container = docker.run(
            image=self.config.image or 'python:3.11-slim',
            detach=True,
            mem_limit=self.config.memory_limit,
            cpu_quota=self.config.cpu_quota,
            network_mode='none' if not self.config.network else 'bridge',
            security_opt=['no-new-privileges:true'],
            read_only=True,
            tmpfs={'/tmp': 'rw,noexec,nosuid,size=100m'},
            user='1000:1000'
        )
    
    async def execute(self, code: str, language: str) -> ExecutionResult:
        if language == 'python':
            cmd = ['python3', '-c', code]
        elif language == 'javascript':
            cmd = ['node', '-e', code]
        elif language == 'bash':
            cmd = ['bash', '-c', code]
        else:
            raise ValueError(f"Unsupported language: {language}")
        
        exec_result = self.container.exec_run(
            cmd,
            demux=True,
            timeout=30
        )
        
        return ExecutionResult(
            exit_code=exec_result.exit_code,
            stdout=exec_result.output[0].decode() if exec_result.output[0] else '',
            stderr=exec_result.output[1].decode() if exec_result.output[1] else ''
        )

class GVisorSandbox:
    """Stronger isolation using gVisor"""
    async def start(self):
        self.container = docker.run(
            image='gcr.io/gvisor/containerd:latest',
            runtime='runsc',
            detach=True,
            # gVisor provides stronger isolation
        )

class FirecrackerSandbox:
    """MicroVM for strongest isolation"""
    async def start(self):
        # Use firecracker-containerd or firecracker directly
        pass

class WasmSandbox:
    """WebAssembly sandbox for safest execution"""
    async def execute(self, code: str, language: str) -> ExecutionResult:
        # Compile to WASM and run in wasmtime/Wasmer
        pass
```

### Permission Model
```python
class PermissionManager:
    def __init__(self):
        self.policies: dict[str, PermissionPolicy] = {}
    
    def define_policy(self, agent_id: str, policy: PermissionPolicy):
        self.policies[agent_id] = policy
    
    async def check_permission(self, agent_id: str, action: Action) -> PermissionResult:
        policy = self.policies.get(agent_id)
        if not policy:
            return PermissionResult(allowed=False, reason="No policy defined")
        
        return await policy.evaluate(action)

class PermissionPolicy:
    def __init__(self):
        self.rules: list[PermissionRule] = []
        self.default_deny = True
    
    def allow(self, action: str, resource: str = '*', conditions: dict = None):
        self.rules.append(PermissionRule(
            action=action, resource=resource, effect='allow', conditions=conditions
        ))
    
    def deny(self, action: str, resource: str = '*', conditions: dict = None):
        self.rules.append(PermissionRule(
            action=action, resource=resource, effect='deny', conditions=conditions
        ))
    
    async def evaluate(self, action: Action) -> PermissionResult:
        for rule in self.rules:
            if rule.matches(action):
                if rule.effect == 'allow':
                    return PermissionResult(allowed=True)
                else:
                    return PermissionResult(allowed=False, reason=rule.reason)
        
        return PermissionResult(allowed=self.default_allow)

# Default policies
DEFAULT_POLICIES = {
    'web_agent': PermissionPolicy(
        rules=[
            PermissionRule(action='browser.*', effect='allow'),
            PermissionRule(action='file.read', resource='~/downloads/*', effect='allow'),
            PermissionRule(action='file.write', resource='~/downloads/*', effect='allow'),
            PermissionRule(action='network.*', resource='api.*', effect='allow'),
            PermissionRule(action='*', resource='*', effect='deny')
        ])
    ],
    'coding_agent': PermissionPolicy(
        rules=[
            PermissionRule(action='file.*', resource='~/projects/*', effect='allow'),
            PermissionRule(action='shell.*', resource='~/projects/*', effect='allow'),
            PermissionRule(action='git.*', effect='allow'),
            PermissionRule(action='network.*', effect='deny'),
        ]
    ),
    'research_agent': PermissionPolicy(
        rules=[
            PermissionRule(action='web.search', effect='allow'),
            PermissionRule(action='web.scrape', effect='allow'),
            PermissionRule(action='file.*', resource='~/research/*', effect='allow'),
            PermissionRule(action='network.*', resource='api.semanticscholar.org', effect='allow'),
        ]
    )
}
```

---

## Remote Access & Collaboration

### Remote Desktop (VNC/RDP)
```python
class RemoteDesktopAgent:
    def __init__(self):
        self.vnc_clients: dict[str, VNCClient] = {}
        self.rdp_clients: dict[str, RDPClient] = {}
    
    async def connect_vnc(self, host: str, port: int = 5900, 
                          password: str = None) -> VNCSession:
        client = VNCClient(host, port, password)
        await client.connect()
        session = VNCSession(client=client)
        self.vnc_clients[session.id] = session
        return session
    
    async def connect_rdp(self, host: str, port: int = 3389,
                          username: str = None, password: str = None) -> RDPSession:
        client = RDPClient(host, port, username, password)
        await client.connect()
        session = RDPSession(client=client)
        self.rdp_clients[session.id] = session
        return session

class VNCSession:
    def __init__(self, client: VNCClient):
        self.client = client
        self.id = str(uuid4())
    
    async def capture(self) -> ScreenCapture:
        return await self.client.capture()
    
    async def send_key(self, key: str):
        await self.client.send_key(key)
    
    async def click(self, x: int, y: int, button: str = 'left'):
        await self.client.click(x, y, button)
    
    async def type_text(self, text: str):
        await self.client.type_text(text)
```

### Collaborative Editing
```python
class CollaborativeEditor:
    def __init__(self):
        self.documents: dict[str, CollaborativeDocument] = {}
    
    async def create_document(self, doc_id: str, content: str = '') -> CollaborativeDocument:
        doc = CollaborativeDocument(doc_id, content)
        self.documents[doc_id] = doc
        return doc
    
    async def join_document(self, doc_id: str, user: User) -> DocumentSession:
        doc = self.documents.get(doc_id)
        if not doc:
            raise ValueError(f"Document {doc_id} not found")
        
        session = DocumentSession(doc, user)
        await doc.join(user)
        return session

class CollaborativeDocument:
    def __init__(self, doc_id: str, content: str):
        self.doc_id = doc_id
        self.content = content
        self.version = 0
        self.operations: list[Operation] = []
        self.cursors: dict[str, Cursor] = {}
        self.lock = asyncio.Lock()
    
    async def apply_operation(self, user_id: str, operation: Operation) -> OperationResult:
        async with self.lock:
            # Transform operation against concurrent ops
            transformed = self.transform_operation(operation)
            
            # Apply
            self.apply_operation(transformed)
            self.operations.append(transformed)
            self.version += 1
            
            # Broadcast to other users
            await self.broadcast(transformed)
            
            return OperationResult(version=self.version, success=True)
    
    def transform_operation(self, op: Operation) -> Operation:
        # Operational Transformation (OT)
        for concurrent_op in self.operations[op.base_version:]:
            op = self.transform(op, concurrent_op)
        return op
```

---

## Observability

### Screen Recording
```python
class ScreenRecorder:
    def __init__(self, output_dir: str = './recordings'):
        self.output_dir = Path(output_dir)
        self.output_dir.mkdir(parents=True, exist_ok=True)
        self.recording: RecordingSession = None
    
    async def start_recording(self, session_id: str, 
                              fps: int = 30, 
                              region: Region = None) -> RecordingSession:
        if self.recording:
            raise RuntimeError("Recording already in progress")
        
        self.recording = RecordingSession(
            id=str(uuid4()),
            session_id=session_id,
            output_path=self.output_dir / f"{session_id}_{datetime.utcnow().isoformat()}.mp4",
            fps=fps,
            region=region
        )
        
        # Start capture loop
        self.recording.task = asyncio.create_task(self._capture_loop())
        return self.recording
    
    async def _capture_loop(self):
        interval = 1.0 / self.recording.fps
        while self.recording and self.recording.active:
            start = time.time()
            frame = await self.capture.capture(self.recording.region)
            await self.recording.writer.write_frame(frame)
            elapsed = time.time() - start
            await asyncio.sleep(max(0, interval - elapsed))
    
    async def stop_recording(self) -> RecordingResult:
        if not self.recording:
            raise RuntimeError("No recording in progress")
        
        self.recording.active = False
        await self.recording.task
        await self.recording.writer.close()
        
        result = RecordingResult(
            path=self.recording.output_path,
            duration=time.time() - self.recording.start_time,
            frames=self.recording.frame_count
        )
        self.recording = None
        return result
```

### Action Logging
```python
class ActionLogger:
    def __init__(self, storage: LogStorage):
        self.storage = storage
        self.buffer: list[ActionLog] = []
        self.flush_interval = 5.0
        self._flush_task = asyncio.create_task(self._periodic_flush())
    
    async def log(self, log: ActionLog):
        self.buffer.append(log)
    
    async def _periodic_flush(self):
        while True:
            await asyncio.sleep(self.flush_interval)
            if self.buffer:
                await self.storage.write_batch(self.buffer)
                self.buffer.clear()
    
    async def log_action(self, agent_id: str, action: Action, 
                         result: ActionResult, context: dict = None):
        log = ActionLog(
            id=str(uuid4()),
            timestamp=datetime.utcnow(),
            agent_id=agent_id,
            action=action,
            result=result,
            context=context or {}
        )
        await self.log(action_log)
    
    async def query_logs(self, query: LogQuery) -> list[ActionLog]:
        return await self.storage.query(query)

class ActionLog:
    def __init__(self, **kwargs):
        for k, v in kwargs.items():
            setattr(self, k, v)
    
    def to_dict(self) -> dict:
        return {k: v for k, v in self.__dict__.items() if not k.startswith('_')}
```

---

## Performance Monitoring

### Agent Metrics
```python
class AgentMetrics:
    def __init__(self, agent_id: str, prometheus: PrometheusClient = None):
        self.agent_id = agent_id
        self.prometheus = prometheus
        
        # Prometheus metrics
        self.request_duration = Histogram(
            'agent_request_duration_seconds',
            'Request duration',
            ['agent', 'operation']
        )
        self.tokens_used = Counter(
            'agent_tokens_total',
            'Tokens used',
            ['agent', 'type']
        )
        self.errors = Counter(
            'agent_errors_total',
            'Total errors',
            ['agent', 'error_type']
        )
        self.active_tasks = Gauge(
            'agent_active_tasks',
            'Active tasks',
            ['agent']
        )
    
    @contextmanager
    def measure(self, operation: str):
        start = time.time()
        self.active_tasks.inc()
        try:
            yield
            self.request_duration.labels(
                agent=self.agent_id, operation=operation
            ).observe(time.time() - start)
        except Exception as e:
            self.errors.labels(
                agent=self.agent_name,
                error_type=type(e).__name__
            ).inc()
            raise
        finally:
            self.active_tasks.dec()
```

### Resource Monitoring
```python
class ResourceMonitor:
    def __init__(self, interval: float = 5.0):
        self.interval = interval
        self.metrics_history: dict[str, list[ResourceMetrics]] = defaultdict(list)
        self._monitor_task = None
    
    async def start(self):
        self._monitor_task = asyncio.create_task(self._monitor_loop())
    
    async def _monitor_loop(self):
        while True:
            metrics = await self.collect_metrics()
            for agent_id, metrics in metrics.items():
                self.metrics_history[agent_id].append(metrics)
                # Keep last 1000 points
                if len(self.metrics_history[agent_id]) > 1000:
                    self.metrics_history[agent_id] = self.metrics_history[agent_id][-1000:]
            await asyncio.sleep(self.interval)
    
    async def collect_metrics(self) -> dict[str, ResourceMetrics]:
        metrics = {}
        for agent_id, agent in self.agents.items():
            process = psutil.Process(agent.pid) if agent.pid else None
            if process:
                metrics[agent_id] = ResourceMetrics(
                    cpu_percent=process.cpu_percent(),
                    memory_mb=process.memory_info().rss / 1024 / 1024,
                    memory_percent=process.memory_percent(),
                    threads=process.num_threads(),
                    open_files=len(process.open_files()),
                    connections=len(process.connections()),
                    timestamp=datetime.utcnow()
                )
        return metrics
```

---

## Related Notes

- [[ToolUse/Function-Calling.md]] — ACI as tool
- [[ToolUse/Browser-Automation.md]] — Browser automation
- [[ToolUse/Code-Execution.md]] — Code execution in sandbox
- [[ToolUse/MCP-Protocol.md]] — MCP for ACI
- [[ToolUse/API-Integration.md]] — API integration
- [[Security/Secure-Coding.md]] — Secure ACI
- [[Security/Supply-Chain-Security.md]] — Supply chain security
- [[Production/Security.md]] — Production security
- [[Production/Reliability.md]] — Reliability patterns
- [[Production/Observability.md]] — Observability
- [[Production/Reliability.md]] — Reliability
- [[Production/Deployment-Architecture.md]] — Deployment architecture
- [[Domains/Coding-Agents.md]] — Coding agents with ACI
- [[Domains/Customer-Facing-Agents.md]] — Customer-facing ACI
- [[Frontiers/Agent-Societies.md]] — Multi-agent ACI
- [[Frontiers/Agent-Societies.md]] — Multi-agent ACI
- [[Frontiers/Research-Directions.md]] — ACI research
- [[Production/Security.md]] — Security hardening
- [[Production/Reliability.md]] — Reliability
- [[Production/Observability.md]] — Observability
- [[Frontiers/Research-Directions.md]] — ACI research

---

*Sources: Playwright, Puppeteer, Selenium, Selenium, CDP, Chrome DevTools Protocol, VNC, RFB, RDP, FreeRDP, X11, Xlib, Quartz, Windows UI Automation, UI Automation, Docker, gVisor, Firecracker, Kata Containers, Kata Containers, Wasm, Wasmtime, Wasmer, WASI, WebAssembly, eBPF, bpftrace, strace, ltrace, perf, SystemTap, Frida, Frida, DynamoRIO, Pin, Intel PT, Intel PT, Intel PT, Frida, DynamoRIO, Pin, Intel PT, QEMU, QEMU, KVM, KVM, Xen, Xen, VirtualBox, VMware, Hyper-V, Parallels, UTM, Limactl, OrbStack, Colima, Rancher Desktop, Docker Desktop, Podman, Podman, Podman, Buildah, Buildah, BuildKit, BuildKit, BuildKit, Kaniko, Kaniko, img, img, img, BuildKit, BuildKit, BuildKit, Kaniko, Kaniko, img, img, img.*