0Pricing
AI Agents · Lesson

Always-On Agent Design Patterns

Background processes, daemon agents, and persistent connection management.

Always-On Agent Design Patterns is a free AI Agents lesson on CoddyKit — lesson 1 of 4. You can read the complete lesson below for free — then practise it hands-on in the browser with a built-in code editor and a 24/7 AI tutor. It is part of the AI Agents learning path, one of 4 lessons in the course, and your progress syncs across the web and the CoddyKit app.

What Is an Always-On Agent?

An always-on agent runs continuously as a background service, waiting for events and taking action proactively. Unlike request-response agents, it persists between interactions and maintains state over time.

Daemon Process Pattern

A daemon process runs in the background, independent of any terminal session. Use Python's daemon threads or a process supervisor to keep the agent running after the terminal closes.

import threading
import time
import signal
import sys

shutdown_flag = threading.Event()

def agent_main_loop():
    print('Agent daemon started')
    while not shutdown_flag.is_set():
        try:
            # Agent work: check for events, process tasks
            perform_agent_cycle()
            shutdown_flag.wait(timeout=60)  # Sleep 60s, wakes on shutdown
        except Exception as e:
            print(f'Agent loop error: {e}')
            shutdown_flag.wait(timeout=5)  # Brief pause on error
    print('Agent daemon stopped')

def perform_agent_cycle():
    print(f'Agent cycle at {time.strftime("%H:%M:%S")}')
    # Check emails, process queue, run scheduled tasks

def handle_signal(signum, frame):
    print(f'Signal {signum} received, shutting down...')
    shutdown_flag.set()

# Register signal handlers for graceful shutdown
signal.signal(signal.SIGTERM, handle_signal)
signal.signal(signal.SIGINT, handle_signal)

# Start as daemon thread
thread = threading.Thread(target=agent_main_loop, daemon=True)
thread.start()
print('Agent running in background')

Watchdog: Auto-Restart on Crash

A watchdog monitors the agent process and restarts it if it crashes. This is essential for production: agents inevitably hit unexpected errors and must recover automatically.

import subprocess
import time
import logging

logger = logging.getLogger('watchdog')

class AgentWatchdog:
    def __init__(self, agent_script: str, max_restarts: int = 10, restart_delay: float = 5.0):
        self.agent_script = agent_script
        self.max_restarts = max_restarts
        self.restart_delay = restart_delay
        self.restart_count = 0
        self.process = None
    
    def start(self):
        self.process = subprocess.Popen(
            ['python', self.agent_script],
            stdout=subprocess.PIPE,
            stderr=subprocess.STDOUT
        )
        logger.info(f'Agent started (PID: {self.process.pid})')
    
    def run_forever(self):
        self.start()
        while True:
            return_code = self.process.wait()
            logger.warning(f'Agent exited with code {return_code}')
            
            if self.restart_count >= self.max_restarts:
                logger.error(f'Max restarts ({self.max_restarts}) reached. Stopping watchdog.')
                break
            
            self.restart_count += 1
            logger.info(f'Restarting agent (attempt {self.restart_count})...')
            time.sleep(self.restart_delay * self.restart_count)  # Backoff
            self.start()

print('AgentWatchdog defined')

Persistent WebSocket Connection

For real-time event delivery, maintain a persistent WebSocket connection. Reconnect automatically if the connection drops — this is the key challenge with always-on agents.

import asyncio
import websockets
import json

async def persistent_websocket_connection(uri: str, on_message):
    backoff = 1
    max_backoff = 60
    
    while True:  # Reconnect forever
        try:
            print(f'Connecting to {uri}')
            async with websockets.connect(uri, ping_interval=30, ping_timeout=10) as ws:
                print('WebSocket connected')
                backoff = 1  # Reset backoff on successful connection
                
                async for raw_message in ws:
                    try:
                        message = json.loads(raw_message)
                        await on_message(message)
                    except json.JSONDecodeError:
                        print(f'Invalid JSON message: {raw_message[:100]}')
        
        except websockets.exceptions.ConnectionClosed as e:
            print(f'WebSocket closed: {e}. Reconnecting in {backoff}s')
        except Exception as e:
            print(f'WebSocket error: {e}. Reconnecting in {backoff}s')
        
        await asyncio.sleep(backoff)
        backoff = min(backoff * 2, max_backoff)  # Exponential backoff

async def handle_ws_message(message: dict):
    print(f'Received: {message}')

print('Persistent WebSocket connection function defined')

Heartbeat Checks

A heartbeat confirms the agent is alive and processing. Send a heartbeat signal every N seconds; if heartbeats stop, the watchdog knows the agent is stuck or dead.

import threading
import time
from datetime import datetime

class HeartbeatMonitor:
    def __init__(self, max_silence_seconds: int = 300):
        self.last_heartbeat = datetime.utcnow()
        self.max_silence = max_silence_seconds
        self.lock = threading.Lock()
    
    def beat(self):
        with self.lock:
            self.last_heartbeat = datetime.utcnow()
    
    def is_alive(self) -> bool:
        with self.lock:
            silence = (datetime.utcnow() - self.last_heartbeat).total_seconds()
            return silence < self.max_silence
    
    def silence_seconds(self) -> float:
        with self.lock:
            return (datetime.utcnow() - self.last_heartbeat).total_seconds()

monitor = HeartbeatMonitor(max_silence_seconds=60)

def agent_with_heartbeat():
    while not shutdown_flag.is_set():
        # Send heartbeat at start of each cycle
        monitor.beat()
        
        # Do agent work
        perform_agent_cycle()
        time.sleep(30)

# External watchdog checks the monitor
def watchdog_check():
    while True:
        if not monitor.is_alive():
            print(f'ALERT: Agent silent for {monitor.silence_seconds():.0f}s')
            # Restart agent here
        time.sleep(30)

print('Heartbeat monitor defined')

Graceful Shutdown

Graceful shutdown completes in-progress work before stopping. It receives a stop signal, prevents new work from starting, finishes current tasks, saves state, and exits cleanly.

import signal
import threading
from contextlib import contextmanager

class GracefulShutdown:
    def __init__(self, timeout: float = 30.0):
        self.should_stop = threading.Event()
        self.active_tasks = 0
        self.lock = threading.Lock()
        self.timeout = timeout
        signal.signal(signal.SIGTERM, self._handle_signal)
        signal.signal(signal.SIGINT, self._handle_signal)
    
    def _handle_signal(self, signum, frame):
        print(f'Shutdown signal received. Waiting for {self.active_tasks} active tasks...')
        self.should_stop.set()
    
    @contextmanager
    def task(self):
        if self.should_stop.is_set():
            raise RuntimeError('Shutdown in progress, not accepting new tasks')
        with self.lock:
            self.active_tasks += 1
        try:
            yield
        finally:
            with self.lock:
                self.active_tasks -= 1
    
    def wait_for_all_tasks(self):
        self.should_stop.wait()
        deadline = time.time() + self.timeout
        while self.active_tasks > 0 and time.time() < deadline:
            time.sleep(0.1)
        if self.active_tasks > 0:
            print(f'WARNING: Forced shutdown with {self.active_tasks} tasks still active')

shutdown = GracefulShutdown(timeout=30)
print('Graceful shutdown manager created')

Handling Disconnects and Reconnects

Always-on agents need strategies for handling service disconnections: buffer events during downtime, replay missed events after reconnect, and avoid losing events that arrived while disconnected.

import asyncio
import redis
from datetime import datetime

r = redis.Redis(host='localhost', port=6379, decode_responses=True)

class DisconnectHandler:
    def __init__(self, buffer_key: str = 'agent:offline_buffer'):
        self.buffer_key = buffer_key
        self.connected = True
    
    def on_disconnect(self):
        self.connected = False
        print(f'Disconnected at {datetime.utcnow()}')
    
    def on_reconnect(self):
        self.connected = True
        print(f'Reconnected at {datetime.utcnow()}')
        self.replay_buffered_events()
    
    def handle_event(self, event: dict):
        if not self.connected:
            # Buffer events for later replay
            import json
            r.lpush(self.buffer_key, json.dumps(event))
            print(f'Event buffered (offline): {event["type"]}')
            return
        self.process_event(event)
    
    def replay_buffered_events(self):
        import json
        replayed = 0
        while True:
            raw = r.rpop(self.buffer_key)
            if not raw:
                break
            event = json.loads(raw)
            self.process_event(event)
            replayed += 1
        if replayed:
            print(f'Replayed {replayed} buffered events')
    
    def process_event(self, event: dict):
        print(f'Processing event: {event["type"]}')

handler = DisconnectHandler()
print('Disconnect handler created')

Process Supervision with systemd

For Linux production deployments, use systemd to manage the agent process. It handles auto-restart, logging to journald, and starts the agent on boot.

# /etc/systemd/system/my-agent.service
SERVICE_FILE = '''
[Unit]
Description=My AI Agent Service
After=network.target

[Service]
Type=simple
User=ubuntu
WorkingDirectory=/home/ubuntu/agent
ExecStart=/home/ubuntu/venv/bin/python agent.py
Restart=always
RestartSec=10
StandardOutput=journal
StandardError=journal

# Environment variables
EnvironmentFile=/home/ubuntu/agent/.env

# Resource limits
MemoryLimit=1G
CPUQuota=50%

[Install]
WantedBy=multi-user.target
'''

# Deploy commands:
# sudo cp my-agent.service /etc/systemd/system/
# sudo systemctl daemon-reload
# sudo systemctl enable my-agent
# sudo systemctl start my-agent
# sudo systemctl status my-agent
# sudo journalctl -u my-agent -f  # Follow logs

print('Systemd service configuration defined')
print('Enables: auto-start on boot, auto-restart on crash, centralized logging')

State Persistence Across Restarts

An always-on agent must save its state so it can resume where it left off after a restart. Save checkpoint data to disk or Redis at regular intervals and after significant state changes.

import json
import os
from datetime import datetime

CHECKPOINT_FILE = '/tmp/agent_checkpoint.json'

def save_checkpoint(state: dict):
    state['last_saved'] = datetime.utcnow().isoformat()
    with open(CHECKPOINT_FILE, 'w') as f:
        json.dump(state, f, indent=2)
    print(f'Checkpoint saved at {state["last_saved"]}')

def load_checkpoint() -> dict:
    if not os.path.exists(CHECKPOINT_FILE):
        print('No checkpoint found, starting fresh')
        return {}
    with open(CHECKPOINT_FILE) as f:
        state = json.load(f)
    print(f'Checkpoint loaded from {state.get("last_saved", "unknown")}')
    return state

# Agent startup
agent_state = load_checkpoint()
last_processed_id = agent_state.get('last_processed_email_id', 0)
print(f'Resuming from email ID: {last_processed_id}')

# After processing each email
agent_state['last_processed_email_id'] = last_processed_id + 1
if agent_state['last_processed_email_id'] % 10 == 0:  # Checkpoint every 10 items
    save_checkpoint(agent_state)

Monitoring Always-On Agents

Track key metrics for always-on agents: uptime, events processed per hour, error rate, memory usage, and last active time. Expose these via a health endpoint or push to a monitoring service.

from fastapi import FastAPI
from datetime import datetime
import psutil
import os

app = FastAPI()
start_time = datetime.utcnow()
events_processed = 0
last_event_time = None

@app.get('/health')
def health_check():
    process = psutil.Process(os.getpid())
    uptime_seconds = (datetime.utcnow() - start_time).total_seconds()
    
    last_active = None
    if last_event_time:
        last_active = (datetime.utcnow() - last_event_time).total_seconds()
    
    return {
        'status': 'ok',
        'uptime_seconds': round(uptime_seconds),
        'events_processed': events_processed,
        'memory_mb': round(process.memory_info().rss / 1024 / 1024, 1),
        'cpu_percent': process.cpu_percent(interval=1),
        'last_event_seconds_ago': round(last_active) if last_active else None,
        'timestamp': datetime.utcnow().isoformat()
    }

Circuit Breaker for External Dependencies

An always-on agent interacts with external services that can fail. A circuit breaker stops making calls to a failing service for a period, preventing cascade failures.

import time
from enum import Enum

class CircuitState(Enum):
    CLOSED = 'closed'      # Normal operation
    OPEN = 'open'          # Service down, not calling
    HALF_OPEN = 'half_open'  # Testing if service recovered

class CircuitBreaker:
    def __init__(self, failure_threshold=5, recovery_timeout=60):
        self.state = CircuitState.CLOSED
        self.failure_count = 0
        self.threshold = failure_threshold
        self.recovery_timeout = recovery_timeout
        self.last_failure_time = None
    
    def call(self, fn, *args):
        if self.state == CircuitState.OPEN:
            if time.time() - self.last_failure_time > self.recovery_timeout:
                self.state = CircuitState.HALF_OPEN
            else:
                raise RuntimeError('Circuit open: service unavailable')
        
        try:
            result = fn(*args)
            if self.state == CircuitState.HALF_OPEN:
                self.state = CircuitState.CLOSED
                self.failure_count = 0
                print('Circuit closed: service recovered')
            return result
        except Exception as e:
            self.failure_count += 1
            self.last_failure_time = time.time()
            if self.failure_count >= self.threshold:
                self.state = CircuitState.OPEN
                print(f'Circuit opened after {self.failure_count} failures')
            raise

circuit = CircuitBreaker(failure_threshold=3, recovery_timeout=30)
print('Circuit breaker created')

Knowledge Check: Always-On Agents

Test your understanding of always-on agent design patterns.

Always-On Design Patterns Summary

Reliable always-on agents combine: daemon processes with signal handling for clean shutdown, watchdogs for automatic restart on crash, persistent WebSocket connections with exponential backoff reconnection, heartbeat checks to detect stuck processes, graceful shutdown to complete in-flight work, and state checkpointing to resume after restart. Use systemd or supervisord for production process management.

Frequently asked questions

Is the “Always-On Agent Design Patterns” lesson free?

Yes — the full text of “Always-On Agent Design Patterns” is free to read here on the web, and the AI Agents course includes 4 lessons in total. To practise it interactively (a built-in code editor and a 24/7 AI tutor) and unlock the rest of the AI Agents course, upgrade to CoddyKit PRO.

What will I learn in “Always-On Agent Design Patterns”?

Background processes, daemon agents, and persistent connection management. You practise AI Agents with hands-on code you run directly in the browser, and a 24/7 AI tutor answers your questions as you work through the lesson.

Do I need any experience to start AI Agents?

No prior experience is required. AI Agents on CoddyKit is structured for beginners through advanced learners; this is — lesson 1 of 4, so you can start here or from the beginning and move at your own pace.

How long does the “Always-On Agent Design Patterns” lesson take?

Most CoddyKit lessons take about 5–10 minutes. Each one is bite-sized and interactive, so you make steady progress and pick up exactly where you left off across the web and the app.

Can I write and run code in this AI Agents lesson?

Yes. Every AI Agents lesson includes a built-in code editor, so you write and run real code right in your browser and get instant AI feedback — no local setup required.

All lessons in this course

  1. Always-On Agent Design Patterns
  2. Proactive Notification and Alert Systems
  3. Context Persistence Across Sessions
  4. Building a Daily Briefing Agent
← Back to AI Agents