Blue-Green Deployments for AI Agents: Zero-Downtime Updates
Deploying updates to AI agents in production is fundamentally different from shipping traditional software. AI agents maintain conversational state, interact with multiple external systems, and often rely on complex chains of reasoning that span several minutes. A naive deployment that simply replaces the running agent can sever active conversations, corrupt in-progress reasoning chains, and degrade user trust. Blue-green deployment offers a disciplined strategy to ship updates without disruption.
What Is Blue-Green Deployment for AI Agents?
Blue-green deployment is a release pattern where two complete, independent environments run side by side. At any given moment, only one environment — referred to as blue or green — serves live traffic. The other environment stands idle, fully provisioned and ready, but receives no production requests until a switch is triggered.
When a new version of the AI agent is ready, you deploy it to the idle environment, run validation checks, and then flip a routing mechanism so all incoming traffic flows to the newly updated environment. The old environment remains available for a defined grace period, allowing in-flight agent sessions to complete naturally before the old environment is decommissioned.
In the context of AI agents, this pattern extends beyond simple HTTP routing. You must also consider:
- Session state migration — active conversations and tool-call chains
- Vector store and embedding consistency — RAG-based agents referencing knowledge bases
- Tool API compatibility — external tools the agent invokes during reasoning
- LLM provider model versioning — underlying models that may change alongside agent logic
Why Zero-Downtime Deployments Matter for AI Agents
AI agents operate in a world of long-running, stateful interactions. A user might spend 20 minutes refining a complex analysis with an agent, iterating through multiple reasoning steps. If a deployment kills that session mid-flight, the user loses context, trust evaporates, and the agent's reliability score plummets. Beyond user experience, there are concrete operational risks:
- Revenue loss — agents handling customer transactions that are interrupted
- Compliance violations — regulated industries where audit trails must remain intact
- Cascading failures — an agent orchestrating multiple microservices that leaves dangling state
- Model inconsistency — half of a reasoning chain running on one model version and half on another
Blue-green deployments mitigate these risks by preserving the old environment long enough for all active sessions to drain. You never run two versions concurrently serving new traffic, so model inconsistency for new sessions is eliminated, while old sessions reach a clean conclusion.
Architecture Overview
A robust blue-green setup for AI agents requires several components working in concert. The diagram below outlines the logical flow, and the subsequent sections provide concrete implementation code.
Key components include:
- Traffic Router / API Gateway — directs incoming requests to the active environment
- Session Store — a shared persistence layer decoupled from compute environments
- Deployment Orchestrator — a script or pipeline that manages the switch sequence
- Health Check Endpoint — a dedicated route on each agent environment for readiness validation
- Graceful Drainage Controller — tracks in-flight sessions and delays teardown
Step 1: Decouple Session State from Compute
The single most important prerequisite is that session state lives outside the agent runtime. Use a shared Redis cluster, a PostgreSQL table, or a dedicated session service. Each agent instance — whether in blue or green — reads and writes sessions from the same external store.
Below is a minimal session store abstraction using Redis. Both blue and green environments connect to the same Redis instance.
import redis
import json
from datetime import datetime, timedelta
from typing import Optional, Dict, Any
class SessionStore:
"""Shared session persistence decoupled from agent compute environments."""
def __init__(self, redis_url: str, ttl_minutes: int = 60):
self.client = redis.from_url(redis_url)
self.ttl = timedelta(minutes=ttl_minutes)
def create_session(self, session_id: str, metadata: Dict[str, Any]) -> None:
key = f"agent:session:{session_id}"
payload = {
"created_at": datetime.utcnow().isoformat(),
"status": "active",
"conversation_history": [],
"tool_call_stack": [],
"metadata": metadata
}
self.client.setex(
key,
int(self.ttl.total_seconds()),
json.dumps(payload)
)
def append_conversation_turn(self, session_id: str, turn: Dict[str, Any]) -> Optional[Dict]:
key = f"agent:session:{session_id}"
raw = self.client.get(key)
if not raw:
return None
session = json.loads(raw)
session["conversation_history"].append(turn)
self.client.setex(key, int(self.ttl.total_seconds()), json.dumps(session))
return session
def get_session(self, session_id: str) -> Optional[Dict[str, Any]]:
raw = self.client.get(f"agent:session:{session_id}")
return json.loads(raw) if raw else None
def mark_completed(self, session_id: str) -> None:
key = f"agent:session:{session_id}"
raw = self.client.get(key)
if raw:
session = json.loads(raw)
session["status"] = "completed"
session["completed_at"] = datetime.utcnow().isoformat()
# Keep completed sessions around for audit for a short window
self.client.setex(key, 300, json.dumps(session))
def count_active_sessions(self) -> int:
"""Returns count of sessions currently in 'active' status."""
count = 0
for key in self.client.scan_iter("agent:session:*"):
raw = self.client.get(key)
if raw:
session = json.loads(raw)
if session.get("status") == "active":
count += 1
return count
Step 2: Build Health-Check Endpoints on Each Agent Environment
Every deployed agent environment must expose a health endpoint that the orchestrator can query before switching traffic. This endpoint should validate internal dependencies — model availability, tool connectivity, vector store reachability — not just process liveness.
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
import asyncio
app = FastAPI()
class HealthStatus(BaseModel):
status: str
version: str
dependencies: dict
active_sessions: int
@app.get("/health/readiness")
async def readiness_check():
"""
Comprehensive readiness probe used by the deployment orchestrator.
Returns 200 only if all subsystems are reachable and functional.
"""
checks = {
"llm_provider": await check_llm_connectivity(),
"vector_store": await check_vector_store(),
"tool_registry": await check_tool_endpoints(),
"session_store": await check_session_store_connectivity(),
}
all_healthy = all(result == "healthy" for result in checks.values())
if not all_healthy:
raise HTTPException(
status_code=503,
detail=f"Readiness check failed: {checks}"
)
return HealthStatus(
status="ready",
version="2.4.1",
dependencies=checks,
active_sessions=0 # New environment starts with zero sessions
)
async def check_llm_connectivity() -> str:
try:
# Ping the model provider with a trivial token-count request
await asyncio.wait_for(provider_ping(), timeout=5.0)
return "healthy"
except Exception:
return "unreachable"
async def check_vector_store() -> str:
try:
# Execute a lightweight similarity search with a known query vector
await asyncio.wait_for(vector_store_ping(), timeout=3.0)
return "healthy"
except Exception:
return "unreachable"
async def check_tool_endpoints() -> str:
try:
# Verify each registered tool endpoint responds
await asyncio.wait_for(check_all_tools(), timeout=8.0)
return "healthy"
except Exception:
return "degraded"
async def check_session_store_connectivity() -> str:
try:
store = get_session_store()
store.client.ping()
return "healthy"
except Exception:
return "unreachable"
# Placeholder implementations — replace with actual connectivity checks
async def provider_ping(): pass
async def vector_store_ping(): pass
async def check_all_tools(): pass
def get_session_store(): return SessionStore("redis://localhost:6379")
Step 3: The Deployment Orchestrator Script
This is the engine that drives the entire blue-green switch. It follows a strict sequence: deploy to idle, validate, drain active sessions from current live environment, flip traffic, and finally decommission the old environment after a safety window.
#!/usr/bin/env python3
"""
Blue-Green Deployment Orchestrator for AI Agent Environments.
Usage: python deploy_orchestrator.py --target-version 2.5.0 --environment-id prod-agents
"""
import argparse
import subprocess
import time
import requests
import json
from datetime import datetime, timedelta
from typing import Literal
class DeploymentOrchestrator:
def __init__(self, environment_id: str, target_version: str):
self.environment_id = environment_id
self.target_version = target_version
self.metadata_store = self._connect_metadata_store()
self.active_color: Literal["blue", "green"] = self._read_current_active()
self.idle_color: Literal["blue", "green"] = "green" if self.active_color == "blue" else "blue"
def _connect_metadata_store(self):
"""Connect to a lightweight metadata store (e.g., Consul KV, etcd, or a DB table)."""
# Simplified: in production, use a proper distributed KV store
return {
"current_active": "blue",
"blue": {"version": "2.4.1", "endpoint": "https://blue-agents.internal"},
"green": {"version": "2.4.0", "endpoint": "https://green-agents.internal"}
}
def _read_current_active(self) -> Literal["blue", "green"]:
return self.metadata_store["current_active"] # type: ignore
def run(self):
"""Execute the full blue-green deployment sequence."""
print(f"=== Blue-Green Deployment: {self.environment_id} -> v{self.target_version} ===")
print(f"Active environment: {self.active_color} | Idle environment: {self.idle_color}")
# Phase 1: Deploy new version to idle environment
self.deploy_to_idle()
# Phase 2: Validate idle environment readiness
self.validate_idle_environment()
# Phase 3: Pre-warm the idle environment
self.prewarm_idle_environment()
# Phase 4: Drain active sessions from current live environment
self.drain_active_sessions()
# Phase 5: Switch traffic to idle environment
self.switch_traffic()
# Phase 6: Monitor new environment post-switch
self.post_switch_monitoring()
# Phase 7: Decommission old environment after safety window
self.decommission_old_environment()
print(f"=== Deployment complete: {self.environment_id} now running v{self.target_version} ===")
def deploy_to_idle(self):
"""Deploy the target version to the idle environment."""
idle_endpoint = self.metadata_store[self.idle_color]["endpoint"]
print(f"[Phase 1] Deploying v{self.target_version} to {self.idle_color} environment at {idle_endpoint}")
# Trigger CI/CD pipeline for the idle environment
result = subprocess.run([
"kubectl", "set", "image",
f"deployment/agent-{self.idle_color}",
f"agent-container=ai-agent:{self.target_version}",
"-n", self.environment_id
], capture_output=True, text=True)
if result.returncode != 0:
raise RuntimeError(f"Deployment failed: {result.stderr}")
# Wait for rollout to complete
subprocess.run([
"kubectl", "rollout", "status",
f"deployment/agent-{self.idle_color}",
"-n", self.environment_id,
"--timeout=300s"
], check=True)
# Update metadata
self.metadata_store[self.idle_color]["version"] = self.target_version
print(f"[Phase 1] Deployment to {self.idle_color} complete")
def validate_idle_environment(self):
"""Run comprehensive health checks against the idle environment."""
idle_endpoint = self.metadata_store[self.idle_color]["endpoint"]
print(f"[Phase 2] Validating {self.idle_color} environment health")
readiness_url = f"{idle_endpoint}/health/readiness"
# Retry loop with backoff — readiness may take time after deployment
max_retries = 10
for attempt in range(1, max_retries + 1):
try:
response = requests.get(readiness_url, timeout=10)
if response.status_code == 200:
health_data = response.json()
print(f" Readiness check passed: status={health_data['status']}, version={health_data['version']}")
# Verify version matches target
if health_data['version'] != self.target_version:
raise RuntimeError(
f"Version mismatch: expected {self.target_version}, got {health_data['version']}"
)
# Verify all dependencies healthy
for dep, dep_status in health_data['dependencies'].items():
if dep_status != "healthy":
raise RuntimeError(f"Dependency {dep} is {dep_status}")
print(f"[Phase 2] All validation checks passed")
return
elif response.status_code == 503:
print(f" Attempt {attempt}: 503 — dependencies not ready yet, retrying...")
else:
print(f" Attempt {attempt}: HTTP {response.status_code}, retrying...")
except requests.RequestException as e:
print(f" Attempt {attempt}: Connection error: {e}, retrying...")
time.sleep(min(5 * attempt, 30)) # Progressive backoff
raise RuntimeError(f"Idle environment failed readiness checks after {max_retries} attempts")
def prewarm_idle_environment(self):
"""Send synthetic traffic to warm caches, JIT compilations, and connection pools."""
idle_endpoint = self.metadata_store[self.idle_color]["endpoint"]
print(f"[Phase 3] Pre-warming {self.idle_color} environment")
warmup_payloads = [
{"prompt": "What is 2 + 2?", "session_id": "warmup-001", "metadata": {"warmup": True}},
{"prompt": "Summarize: AI deployment patterns", "session_id": "warmup-002", "metadata": {"warmup": True}},
{"prompt": "List three colors", "session_id": "warmup-003", "metadata": {"warmup": True}},
]
for payload in warmup_payloads:
try:
resp = requests.post(
f"{idle_endpoint}/agent/query",
json=payload,
timeout=30
)
print(f" Warmup query {payload['session_id']}: HTTP {resp.status_code}")
except Exception as e:
print(f" Warmup query failed (non-fatal): {e}")
print(f"[Phase 3] Pre-warming complete")
def drain_active_sessions(self):
"""Wait for active sessions on the current live environment to complete."""
active_endpoint = self.metadata_store[self.active_color]["endpoint"]
print(f"[Phase 4] Draining active sessions from {self.active_color} environment")
drain_deadline = datetime.utcnow() + timedelta(minutes=15) # Max drain window
check_interval = 10 # seconds
while datetime.utcnow() < drain_deadline:
try:
# Query session count from the active environment's health endpoint
resp = requests.get(f"{active_endpoint}/health/readiness", timeout=5)
if resp.status_code == 200:
health = resp.json()
active_count = health.get("active_sessions", 0)
print(f" Active sessions remaining: {active_count}")
if active_count == 0:
print(f"[Phase 4] Drain complete — all sessions concluded")
return
except Exception as e:
print(f" Warning: could not query session count: {e}")
time.sleep(check_interval)
# If we reach here, some sessions didn't drain in time
print(f"[Phase 4] Drain timeout reached — forcing switch with {self._get_remaining_sessions()} remaining sessions")
# These sessions will continue on the old environment until natural completion
# (old environment stays alive post-switch for this purpose)
def _get_remaining_sessions(self) -> int:
try:
active_endpoint = self.metadata_store[self.active_color]["endpoint"]
resp = requests.get(f"{active_endpoint}/health/readiness", timeout=5)
return resp.json().get("active_sessions", 0) if resp.status_code == 200 else -1
except Exception:
return -1
def switch_traffic(self):
"""Atomically flip the traffic router to point to the new environment."""
print(f"[Phase 5] Switching traffic from {self.active_color} to {self.idle_color}")
idle_endpoint = self.metadata_store[self.idle_color]["endpoint"]
# Update the traffic router configuration
# This example uses an API Gateway config update; in practice this could be:
# - Nginx/Envoy configuration reload
# - Kubernetes Service selector update
# - DNS weighted routing update
# - Load balancer target group swap
result = subprocess.run([
"kubectl", "patch", "service", "agent-gateway",
"-n", self.environment_id,
"--type=merge",
"-p", json.dumps({
"spec": {
"selector": {
"deployment": f"agent-{self.idle_color}"
}
}
})
], capture_output=True, text=True)
if result.returncode != 0:
raise RuntimeError(f"Traffic switch failed: {result.stderr}")
# Update metadata to reflect new active environment
old_active = self.active_color
self.metadata_store["current_active"] = self.idle_color
self.active_color = self.idle_color
self.idle_color = old_active
print(f"[Phase 5] Traffic switched — new active environment is {self.active_color}")
def post_switch_monitoring(self):
"""Monitor the new active environment for errors post-switch."""
print(f"[Phase 6] Post-switch monitoring of {self.active_color} environment")
active_endpoint = self.metadata_store[self.active_color]["endpoint"]
monitoring_duration = 120 # seconds
check_interval = 5
checks_total = monitoring_duration // check_interval
failures = 0
for i in range(checks_total):
try:
resp = requests.get(f"{active_endpoint}/health/readiness", timeout=5)
if resp.status_code != 200:
failures += 1
print(f" Check {i+1}/{checks_total}: FAILURE — HTTP {resp.status_code}")
else:
health = resp.json()
session_count = health.get("active_sessions", 0)
print(f" Check {i+1}/{checks_total}: OK — {session_count} active sessions")
except Exception as e:
failures += 1
print(f" Check {i+1}/{checks_total}: EXCEPTION — {e}")
time.sleep(check_interval)
failure_rate = failures / checks_total
print(f" Post-switch failure rate: {failure_rate:.2%}")
if failure_rate > 0.2: # More than 20% failures — rollback
print(f"[Phase 6] CRITICAL: High failure rate detected. Initiating rollback...")
self.rollback()
raise RuntimeError("Rollback executed due to high post-switch failure rate")
print(f"[Phase 6] Post-switch monitoring passed")
def rollback(self):
"""Immediately switch traffic back to the previous active environment."""
print(" ROLLBACK: Reverting traffic to previous active environment")
previous_active = "green" if self.active_color == "blue" else "blue"
previous_endpoint = self.metadata_store[previous_active]["endpoint"]
subprocess.run([
"kubectl", "patch", "service", "agent-gateway",
"-n", self.environment_id,
"--type=merge",
"-p", json.dumps({
"spec": {"selector": {"deployment": f"agent-{previous_active}"}}
})
], check=True)
self.metadata_store["current_active"] = previous_active
print(f" ROLLBACK complete — active environment restored to {previous_active}")
def decommission_old_environment(self):
"""Decommission the old environment after a safety window."""
old_color = "green" if self.active_color == "blue" else "blue"
print(f"[Phase 7] Scheduling decommission of {old_color} environment")
safety_window_minutes = 30
print(f" Waiting {safety_window_minutes} minutes safety window...")
time.sleep(60) # Simulated; in production, use a scheduled job
# Final check: ensure zero sessions remain on old environment
old_endpoint = self.metadata_store[old_color]["endpoint"]
try:
resp = requests.get(f"{old_endpoint}/health/readiness", timeout=5)
if resp.status_code == 200:
remaining = resp.json().get("active_sessions", 0)
if remaining > 0:
print(f" Warning: {remaining} sessions still active on {old_color} — delaying teardown")
return # Don't decommission yet
except Exception:
print(f" Old environment unreachable — safe to decommission")
# Scale down old deployment
subprocess.run([
"kubectl", "scale", "deployment", f"agent-{old_color}",
"-n", self.environment_id,
"--replicas=0"
], check=True)
print(f"[Phase 7] {old_color} environment decommissioned")
if __name__ == "__main__":
parser = argparse.ArgumentParser(description="Blue-Green Deployment Orchestrator")
parser.add_argument("--target-version", required=True, help="Version to deploy")
parser.add_argument("--environment-id", required=True, help="Environment identifier")
args = parser.parse_args()
orchestrator = DeploymentOrchestrator(args.environment_id, args.target_version)
orchestrator.run()
Step 4: API Gateway Traffic Routing Configuration
The traffic router is the linchpin of blue-green deployments. Below is an Nginx configuration that routes traffic to the active environment based on a shared state file, which the orchestrator updates during the switch phase.
# nginx.conf — Blue-Green Traffic Router for AI Agent Environments
# The orchestrator updates /etc/nginx/active-upstream.conf during the switch phase
worker_processes auto;
error_log /var/log/nginx/error.log warn;
events {
worker_connections 4096;
}
http {
upstream agent_upstream {
# The active environment endpoint is managed by the orchestrator
# Format: server : max_conns=;
include /etc/nginx/active-upstream.conf;
}
# Websocket upstream for persistent agent sessions
upstream agent_ws_upstream {
include /etc/nginx/active-upstream-ws.conf;
}
server {
listen 443 ssl;
server_name agents.example.com;
ssl_certificate /etc/ssl/agents-cert.pem;
ssl_certificate_key /etc/ssl/agents-key.pem;
# Health check endpoint — proxies to active environment
location /health/ {
proxy_pass http://agent_upstream;
proxy_http_version 1.1;
proxy_set_header Connection "";
proxy_read_timeout 5s;
}
# Main agent query endpoint — supports long-running requests
location /agent/ {
proxy_pass http://agent_upstream;
proxy_http_version 1.1;
proxy_set_header Connection "";
proxy_set_header X-Real-IP $remote_addr;
proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
proxy_set_header X-Environment-ID $env_id;
# Extended timeouts for AI agent reasoning
proxy_read_timeout 300s;
proxy_send_timeout 300s;
# Retry only once, and only on the same upstream
proxy_next_upstream error timeout;
proxy_next_upstream_tries 1;
}
# WebSocket endpoint for streaming agent responses
location /agent/stream/ {
proxy_pass http://agent_ws_upstream;
proxy_http_version 1.1;
proxy_set_header Upgrade $http_upgrade;
proxy_set_header Connection "upgrade";
proxy_set_header Host $host;
proxy_read_timeout 600s; # 10 minutes for streaming sessions
}
# Admin endpoint for orchestrator to query session counts
location /admin/sessions/count {
proxy_pass http://agent_upstream;
proxy_http_version 1.1;
proxy_set_header Connection "";
proxy_read_timeout 5s;
# Restrict to internal IPs in production
allow 10.0.0.0/8;
deny all;
}
}
}
The orchestrator updates the active upstream configuration file atomically using a symlink swap or file rename, which Nginx picks up on the next reload signal:
#!/bin/bash
# switch-upstream.sh — called by the orchestrator during traffic switch phase
# Usage: ./switch-upstream.sh blue|green
TARGET_COLOR=$1
if [ "$TARGET_COLOR" = "blue" ]; then
echo "server blue-agents.internal:8080 max_conns=500;" > /etc/nginx/active-upstream.conf.new
echo "server blue-agents.internal:8081 max_conns=500;" > /etc/nginx/active-upstream-ws.conf.new
elif [ "$TARGET_COLOR" = "green" ]; then
echo "server green-agents.internal:8080 max_conns=500;" > /etc/nginx/active-upstream.conf.new
echo "server green-agents.internal:8081 max_conns=500;" > /etc/nginx/active-upstream-ws.conf.new
else
echo "Invalid color: $TARGET_COLOR" >&2
exit 1
fi
# Atomically replace the upstream configuration files
mv /etc/nginx/active-upstream.conf.new /etc/nginx/active-upstream.conf
mv /etc/nginx/active-upstream-ws.conf.new /etc/nginx/active-upstream-ws.conf
# Gracefully reload Nginx — zero-downtime config application
nginx -s reload
echo "Traffic switched to $TARGET_COLOR environment"
Step 5: Graceful Session Drainage on the Agent Side
Each agent instance must cooperate with the drainage process. When the orchestrator signals that a switch is imminent, the agent runtime should stop accepting new sessions while allowing existing sessions to complete naturally.
import asyncio
import signal
from datetime import datetime
class GracefulDrainageHandler:
"""
Mixin for agent runtimes that supports graceful drainage.
When drainage mode is activated, new session requests receive
a redirect response, while existing sessions run to completion.
"""
def __init__(self):
self.draining = False
self.active_session_ids: set = set()
self._drain_start_time: datetime | None = None
self._max_drain_seconds = 900 # 15 minutes max
def activate_drainage(self) -> dict:
"""Called by orchestrator via admin endpoint to begin drainage."""
self.draining = True
self._drain_start_time = datetime.utcnow()
return {
"status": "draining",
"active_sessions": len(self.active_session_ids),
"drain_started_at": self._drain_start_time.isoformat(),
"max_drain_seconds": self._max_drain_seconds
}
def check_drainage_complete(self) -> bool:
"""Returns True when all sessions have drained or timeout reached."""
if not self.draining:
return True
elapsed = (datetime.utcnow() - self._drain_start_time).total_seconds()
if elapsed > self._max_drain_seconds:
return True # Force-complete after timeout
return len(self.active_session_ids) == 0
async def handle_incoming_request(self, request):
"""Wrapper around request handling that respects drainage mode."""
session_id = request.get("session_id")
if self.draining and not session_id:
# New session request during drainage — redirect to active environment
return {
"status": "redirect",
"message": "Environment is draining. Please retry.",
"retry_after_seconds": 5,
"active_environment": self._get_active_env_endpoint()
}
if session_id and session_id in self.active_session_ids:
# Existing session — process normally
self.active_session_ids.add(session_id)
try:
result = await self.process_request(request)
return result
finally:
# Don't remove from active set until session fully completes
pass
if session_id and session_id not in self.active_session_ids:
if self.draining:
return {
"status": "redirect",
"message": "Environment is draining. Please retry.",
"retry_after_seconds": 5
}
# Normal new session handling
self.active_session_ids.add(session_id)
try:
result = await self.process_request(request)
return result
finally:
self.active_session_ids.discard(session_id)
def _get_active_env_endpoint(self) -> str:
"""Query metadata store for the current active environment endpoint."""
# In production, query Consul/etcd or receive via orchestrator callback
return "https://active-agents.internal"
async def process_request(self, request):
"""Placeholder for actual agent request processing."""
await asyncio.sleep(1)
return {"response": "processed"}
Best Practices for Blue-Green AI Agent Deployments
1. Maintain Backward-Compatible Session Schemas
When the session store schema changes between versions, the old environment must still be able to read sessions written by the new environment during the drainage overlap period. Use schema versioning and migration-tolerant deserialization:
import json
from typing import Dict, Any
class SchemaVersionedSessionStore:
"""Session store that handles schema evolution across blue/green versions."""
CURRENT_SCHEMA_VERSION = 3
@staticmethod
def deserialize_session(raw: str) -> Dict[str, Any]:
session = json.loads(raw)
schema_version = session.get("schema_version", 1)
# Migrate older schemas to current on read
if schema_version < SchemaVersionedSessionStore.CURRENT_SCHEMA_VERSION:
session = SchemaVersionedSessionStore._migrate(session, schema_version)
return session
@staticmethod
def _migrate(session: Dict, from_version: int) -> Dict:
"""Apply sequential migrations to bring session to current schema."""
migrated = session.copy()
if from_version < 2:
# v1 -> v2: conversation_history changed from list of strings to list of dicts
if "conversation_history" in migrated:
old_history = migrated["conversation_history"]
if old_history and isinstance(old_history[0], str):
migrated["conversation_history"] = [
{"role": "user" if i % 2 == 0 else "assistant", "content": msg}
for i, msg in enumerate(old_history)
]
if from_version < 3:
# v2 -> v3: added tool_call_stack with execution metadata
if "tool_call_stack" not in migrated:
migrated["tool_call_stack"] = []
migrated["schema_version"] = SchemaVersionedSessionStore.CURRENT_SCHEMA_VERSION
return migrated
2. Run Smoke Tests Against the Idle Environment Before Switching
Beyond health checks, execute end-to-end smoke tests that simulate real agent interactions. These should cover the most critical paths: simple Q&A, tool invocation, streaming responses, and session continuation.
#!/usr/bin/env python3
"""Smoke test suite for pre-switch validation of idle environment."""
import requests
import json
import sys
IDLE_ENDPOINT = sys.argv[1] if len(sys.argv) > 1 else "https://green-agents.internal"
def test_basic_query():
"""Verify the agent can handle a simple prompt."""
resp = requests.post(f"{IDLE_ENDPOINT}/agent/query", json={
"prompt": "What is the capital of France?",
"session_id": "smoke-test-001",
"metadata": {"test": True}
}, timeout=30)
assert resp.status_code == 200, f"Basic query failed: {resp.status_code}"
data = resp.json()
assert "response" in data, "Response missing 'response' field"
assert "Paris" in data["response"] or "paris" in data["response"].lower(), \
f"Expected 'Paris' in response, got: {data['response'][:100]}"
print(" [PASS] basic_query")
def test_tool_invocation():
"""Verify agent can invoke registered tools."""
resp = requests.post(f"{IDLE_ENDPOINT}/agent/query", json={
"prompt": "Calculate