Building 24/7 Autonomous Agents: Engineering Constraints and Mercury Patterns

class BoundedState:
    def __init__(self, max_active_tokens=1000, summary_model="gpt-4o-mini"):
        self.active_buffer = []
        self.historical_summary = ""
        self.max_tokens = max_active_tokens
        self.summarizer = SummarizerClient(model=summary_model)

    def add_interaction(self, user_input, agent_response):
        # Tokenize and check length
        input_tokens = count_tokens(user_input)
        response_tokens = count_tokens(agent_response)
        
        # Pre-check: If adding this would exceed limit, trigger compaction
        current_load = self.get_current_load() 
        if current_load + input_tokens + response_tokens > self.max_tokens:
            self._compact()
            
        self.active_buffer.append({"in": user_input, "out": agent_response})
        
    def _compact(self):
        """
        Move the oldest 50% of active_buffer into historical_summary.
        """
        # Split buffer
        split_index = len(self.active_buffer) // 2
        old_interactions = self.active_buffer[:split_index]
        
        # Generate summary of old interactions
        summary_prompt = f"Summarize the key facts and decisions: {old_interactions}"
        new_summary = self.summarizer.generate(summary_prompt, self.historical_summary)
        
        # Update state
        self.historical_summary = new_summary
        self.active_buffer = self.active_buffer[split_index:]

// Pseudo-code for the Constraint Engine
import { ZodSchema } from 'zod';

const ActionSchema = z.object({
  tool_name: z.enum(['query_db', 'send_email', 'update_config']),
  parameters: z.object({
    // Strict parameter validation
    query: z.string().max(1000),
    recipient: z.string().email(),
    config_key: z.string().regex(/^[a-z_]+$/),
  })
});

export function executeAction(rawLLMOutput: string, context: AgentContext): void {
  try {
    // 1. Parse and Validate
    const validatedAction = ActionSchema.parse(rawLLMOutput);
    
    // 2. Permission Check
    if (!context.hasPermission(validatedAction.tool_name)) {
      throw new PermissionError(`Tool ${validatedAction.tool_name} not authorized for this tier.`);
    }

    // 3. Cost Check (Predictive)
    const estimatedCost = calculateCost(validatedAction);
    if (context.budget.exhausted(estimatedCost)) {
      throw new BudgetExhaustionError("Action would exceed monthly budget.");
    }

    // 4. Execute
    const result = toolRegistry.get(validatedAction.tool_name).run(validatedAction.parameters);
    
    // 5. Feedback
    context.state.logAction(validatedAction, result);
    
  } catch (error) {
    // Feed error back to LLM as a "System Constraint Violation"
    context.state.setErrorMessage(`Constraint Violation: ${error.message}. Please adjust your plan.`);
  }
}

## SYSTEM STATUS (Real-Time)
- **Budget Remaining**: 12% of monthly cap.
- **Database Latency**: High (>500ms). Avoid complex queries.
- **Current Time**: 04:00 UTC (Low traffic).

**Directive**: Due to high database latency, prioritize caching logic. Do not execute writes. Check cache hit ratio before deciding next action.

This 'adaptive throttling' mechanism adjusts the agent's action frequency based on real‑time load metrics such as CPU utilization, queue depth, or downstream latency. By treating the agent as a proportional‑integral controller, we can smooth bursts and avoid thundering‑herd problems.

### Heartbeat & Health Checks
A 24/7 agent must know when it is alive and when its dependencies are healthy. The simplest heartbeat is a periodic timestamp written to a shared store (e.g., Redis, etcd, or a database table). Each dependent service reads the timestamp and treats the agent as unavailable if the age exceeds a threshold. Health checks go a step further: they probe critical external APIs, verify circuit‑breaker states, and expose a `/health` endpoint that returns `200` only when all subsystems report OK. Combining heartbeat liveness with readiness probes lets orchestration platforms (Kubernetes, Nomad) restart the agent only when it is truly stuck, not merely slow.

### State Reconciliation & Idempotency
Autonomous agents often operate on eventually‑consistent state. To prevent drift, each action includes a deterministic *action ID* derived from the input payload (e.g., a hash of the request). Before applying an action, the agent checks a persistent log of processed IDs. If the ID exists, the action is skipped; otherwise, it is executed and the ID is appended to the log. This idempotency layer makes retries safe and enables exactly‑once semantics even when the underlying transport guarantees at‑least‑once delivery.

### Backpressure Propagation
When downstream services saturate, the agent should not blindly increase its request rate. Instead, it exposes a *backpressure signal*—such as the length of an internal work queue or the latency of the last N calls—and feeds that signal back into its adaptive throttling controller. If the signal exceeds a configured limit, the controller increases the sleep interval multiplicatively (e.g., `sleep *= 1.5`) until the signal falls below the threshold. Conversely, when the signal is low, the interval decays slowly (e.g., `sleep *= 0.9`) to recover throughput without overshooting.

### Self‑Healing Restart
Long‑running agents can accumulate memory leaks or stuck threads. A lightweight watchdog runs in a separate thread or process and monitors key health metrics (CPU, memory, garbage‑collection pauses). If any metric crosses a safety threshold for a configurable duration, the watchdog triggers a graceful shutdown: it stops accepting new work, drains the queue, persists any checkpoint, and exits. The orchestration layer then restarts the agent with a clean state. This pattern isolates catastrophic failures without affecting the rest of the system.

## Runnable Code Example (Python 3.11)

Below is a minimal, production‑ready skeleton that puts the Mercury Patterns together. It uses only the standard library and `asyncio`.

```python
import asyncio
import time
import hashlib
import os
from collections import deque
from typing import Awaitable, Callable

# --------------------------
# Configuration (tune per env)
# --------------------------
HEARTBEAT_INTERVAL = 5.0          # seconds
HEALTH_CHECK_INTERVAL = 15.0
MAX_QUEUE_SIZE = 100
TARGET_LATENCY_MS = 200
ADAPTIVE_KP = 0.1                 # proportional gain
ADAPTIVE_KI = 0.01                # integral gain
SLEEP_MIN = 0.05                  # seconds
SLEEP_MAX = 5.0

# --------------------------
# Shared state
# --------------------------
work_queue: asyncio.Queue = asyncio.Queue(maxsize=MAX_QUEUE_SIZE)
processed_ids: set[str] = set()
latency_history: deque[float] = deque(maxlen=20)
sleep_interval = SLEEP_MIN
integral_error = 0.0
shutdown_event = asyncio.Event()


async def heartbeat(stop: asyncio.Event) -> None:
    """Write a timestamp to a file every HEARTBEAT_INTERVAL seconds."""
    path = "/tmp/agent_heartbeat.txt"
    while not stop.is_set():
        with open(path, "w") as f:
            f.write(str(time.time()))
        await asyncio.sleep(HEARTBEAT_INTERVAL)


async def health_check(stop: asyncio.Event) -> bool:
    """Stub: replace with real dependency probes."""
    await asyncio.sleep(0)  # yield
    return True  # assume healthy


async def adaptive_controller() -> None:
    """Adjust sleep_interval based on observed latency."""
    global sleep_interval, integral_error
    while not shutdown_event.is_set():
        await asyncio.sleep(1.0)  # control loop period
        if not latency_history:
            continue
        avg_latency = sum(latency_history) / len(latency_history)
        error = TARGET_LATENCY_MS - avg_latency  # positive => we can go faster
        integral_error += error
        # PI controller
        adjustment = ADAPTIVE_KP * error + ADAPTIVE_KI * integral_error
        sleep_interval = max(SLEEP_MIN, min(SLEEP_MAX, sleep_interval + adjustment * 0.01))


async def worker() -> None:
    """Consume work items, enforce idempotency, and record latency."""
    while not shutdown_event.is_set():
        try:
            payload = await asyncio.wait_for(work_queue.get(), timeout=1.0)
        except asyncio.TimeoutError:
            continue

        action_id = hashlib.sha256(payload.encode()).hexdigest()
        if action_id in processed_ids:
            work_queue.task_done()
            continue

        start = time.time()
        try:
            # Replace with actual work: call external API, DB write, etc.
            await asyncio.sleep(0.05)  # simulated work
        finally:
            latency = (time.time() - start) * 1000
            latency_history.append(latency)
            processed_ids.add(action_id)
            work_queue.task_done()


async def producer() -> None:
    """Generate synthetic work; in a real system this would be an event source."""
    i = 0
    while not shutdown_event.is_set():
        await work_queue.put(f"task-{i}")
        i += 1
        await asyncio.sleep(0.01)  # produce faster than we can consume to test backpressure


async def main() -> None:
    stop_event = asyncio.Event()
    # Launch all concerns as independent tasks
    tasks = [
        asyncio.create_task(heartbeat(stop_event)),
        asyncio.create_task(adaptive_controller()),
        asyncio.create_task(worker()),
        asyncio.create_task(producer()),
    ]

    try:
        # Run until a termination signal (e.g., SIGTERM) is received
        await asyncio.sleep(3600)  # placeholder for real signal handling
    finally:
        stop_event.set()
        shutdown_event.set()
        await asyncio.gather(*tasks, return_exceptions=True)


if __name__ == "__main__":
    try:
        asyncio.run(main())
    except KeyboardInterrupt:
        pass