Data Retention and Archival for AI Agent Systems: Compliance-Ready Data Lifecycle

Why AI Agent Data Needs Lifecycle Management

AI agents accumulate data fast. Every conversation, tool call, retrieved document, and user interaction generates records. Without a data lifecycle strategy, storage costs grow unbounded, regulatory exposure increases with every record retained beyond its useful life, and deletion requests from users become engineering emergencies instead of routine operations.

A compliance-ready data lifecycle system enforces retention policies automatically, archives data that is no longer active but must be kept, purges data that has exceeded its retention period, and handles right-to-deletion requests within regulatory timelines.

Defining Retention Policies

Different data types have different retention requirements. Conversation logs might be kept for 90 days active, then archived for 2 years. PII-containing records have shorter active periods. Financial transaction data might need 7-year retention.

from dataclasses import dataclass
from enum import Enum
from datetime import datetime, timedelta
from typing import Optional, List, Dict

class RetentionAction(str, Enum):
    KEEP = "keep"
    ARCHIVE = "archive"
    DELETE = "delete"

class DataCategory(str, Enum):
    CONVERSATION = "conversation"
    USER_PROFILE = "user_profile"
    FEEDBACK = "feedback"
    ANALYTICS = "analytics"
    AUDIT_LOG = "audit_log"
    PII = "pii"

@dataclass
class RetentionPolicy:
    category: DataCategory
    active_days: int
    archive_days: int
    description: str

    def get_action(self, created_at: datetime) -> RetentionAction:
        age = datetime.utcnow() - created_at
        if age <= timedelta(days=self.active_days):
            return RetentionAction.KEEP
        elif age <= timedelta(
            days=self.active_days + self.archive_days
        ):
            return RetentionAction.ARCHIVE
        return RetentionAction.DELETE

class PolicyRegistry:
    def __init__(self):
        self.policies: Dict[DataCategory, RetentionPolicy] = {}

    def register(self, policy: RetentionPolicy):
        self.policies[policy.category] = policy

    def get_policy(self, category: DataCategory) -> RetentionPolicy:
        if category not in self.policies:
            raise ValueError(f"No policy for category: {category}")
        return self.policies[category]

# Example configuration
registry = PolicyRegistry()
registry.register(RetentionPolicy(
    category=DataCategory.CONVERSATION,
    active_days=90,
    archive_days=730,
    description="Conversations: 90 days active, 2 years archived",
))
registry.register(RetentionPolicy(
    category=DataCategory.PII,
    active_days=30,
    archive_days=0,
    description="PII: 30 days then permanent deletion",
))
registry.register(RetentionPolicy(
    category=DataCategory.AUDIT_LOG,
    active_days=365,
    archive_days=2555,
    description="Audit logs: 1 year active, 7 years archived",
))

Archival Engine

The archival engine moves data from active storage to cold storage while preserving the ability to retrieve it for audits or legal holds.

import json
import gzip
from pathlib import Path
from typing import AsyncIterator

class ArchivalEngine:
    def __init__(self, archive_path: str, db_pool):
        self.archive_path = Path(archive_path)
        self.archive_path.mkdir(parents=True, exist_ok=True)
        self.db_pool = db_pool

    async def archive_conversations(
        self, before_date: datetime
    ) -> int:
        async with self.db_pool.acquire() as conn:
            rows = await conn.fetch("""
                SELECT id, messages, metadata, created_at
                FROM conversations
                WHERE created_at < $1 AND archived = FALSE
                LIMIT 1000
            """, before_date)

        if not rows:
            return 0

        # Write to compressed archive files grouped by month
        grouped = {}
        for row in rows:
            month_key = row["created_at"].strftime("%Y-%m")
            if month_key not in grouped:
                grouped[month_key] = []
            grouped[month_key].append({
                "id": str(row["id"]),
                "messages": row["messages"],
                "metadata": row["metadata"],
                "created_at": row["created_at"].isoformat(),
            })

        for month_key, records in grouped.items():
            archive_file = (
                self.archive_path / f"conversations_{month_key}.jsonl.gz"
            )
            mode = "ab" if archive_file.exists() else "wb"
            with gzip.open(archive_file, mode) as f:
                for record in records:
                    line = json.dumps(record) + "\n"
                    f.write(line.encode())

        # Mark as archived in database
        async with self.db_pool.acquire() as conn:
            ids = [row["id"] for row in rows]
            await conn.execute("""
                UPDATE conversations SET archived = TRUE
                WHERE id = ANY($1)
            """, ids)

        return len(rows)

    async def retrieve_archived(
        self, conversation_id: str
    ) -> Optional[dict]:
        for archive_file in self.archive_path.glob("*.jsonl.gz"):
            with gzip.open(archive_file, "rt") as f:
                for line in f:
                    record = json.loads(line)
                    if record["id"] == conversation_id:
                        return record
        return None

GDPR Right-to-Deletion Handler

When a user requests deletion, every trace of their data must be removed from active storage, archives, vector databases, and logs within the regulatory timeline (typically 30 days for GDPR).

@dataclass
class DeletionRequest:
    request_id: str
    user_id: str
    requested_at: datetime
    deadline: datetime
    status: str = "pending"
    deletion_log: List[str] = None

    def __post_init__(self):
        if self.deletion_log is None:
            self.deletion_log = []

class GDPRDeletionHandler:
    def __init__(self, db_pool, archive_engine, vector_store):
        self.db_pool = db_pool
        self.archive_engine = archive_engine
        self.vector_store = vector_store

    async def process_deletion(
        self, request: DeletionRequest
    ) -> DeletionRequest:
        # Stage 1: Delete from active database
        async with self.db_pool.acquire() as conn:
            result = await conn.execute("""
                DELETE FROM conversations
                WHERE user_id = $1
            """, request.user_id)
            request.deletion_log.append(
                f"Deleted {result} active conversations"
            )

            result = await conn.execute("""
                DELETE FROM user_profiles
                WHERE user_id = $1
            """, request.user_id)
            request.deletion_log.append(
                f"Deleted {result} user profile records"
            )

            result = await conn.execute("""
                DELETE FROM feedback_events
                WHERE conversation_id IN (
                    SELECT id FROM conversations
                    WHERE user_id = $1
                )
            """, request.user_id)
            request.deletion_log.append(
                f"Deleted {result} feedback events"
            )

        # Stage 2: Delete from vector store
        deleted_vectors = await self.vector_store.delete_by_metadata(
            {"user_id": request.user_id}
        )
        request.deletion_log.append(
            f"Deleted {deleted_vectors} vector embeddings"
        )

        # Stage 3: Record the deletion for audit trail
        async with self.db_pool.acquire() as conn:
            await conn.execute("""
                INSERT INTO deletion_audit_log
                    (request_id, user_id, completed_at, actions)
                VALUES ($1, $2, $3, $4)
            """,
                request.request_id,
                request.user_id,
                datetime.utcnow(),
                json.dumps(request.deletion_log),
            )

        request.status = "completed"
        return request

Automated Lifecycle Runner

A scheduled job that enforces all retention policies automatically.

import logging

logger = logging.getLogger(__name__)

class LifecycleRunner:
    def __init__(self, registry, archive_engine, db_pool):
        self.registry = registry
        self.archive_engine = archive_engine
        self.db_pool = db_pool

    async def run(self):
        for category, policy in self.registry.policies.items():
            archive_before = datetime.utcnow() - timedelta(
                days=policy.active_days
            )
            delete_before = datetime.utcnow() - timedelta(
                days=policy.active_days + policy.archive_days
            )

            archived = await self.archive_engine.archive_conversations(
                before_date=archive_before
            )
            logger.info(
                f"[{category.value}] Archived {archived} records"
            )

            if policy.archive_days > 0:
                deleted = await self._purge_old_archives(
                    delete_before
                )
                logger.info(
                    f"[{category.value}] Purged {deleted} "
                    f"expired archives"
                )

    async def _purge_old_archives(self, before: datetime) -> int:
        async with self.db_pool.acquire() as conn:
            result = await conn.execute("""
                DELETE FROM conversations
                WHERE archived = TRUE AND created_at < $1
            """, before)
        return int(result.split()[-1])

FAQ

How do I handle legal holds that override retention policies?

Implement a legal hold flag on records that prevents the lifecycle runner from archiving or deleting them. When legal places a hold on a matter, mark all related conversations and user records with a hold ID. The lifecycle runner checks for active holds before any deletion. Only release records for normal lifecycle processing after legal explicitly lifts the hold.

Should I delete data from backups too for GDPR compliance?

GDPR regulators generally accept that backup deletion is impractical if you have documented procedures showing the data will be deleted when the backup expires through its normal rotation schedule. Document your backup retention period, and ensure deleted data is not restored from backups. If your backup retention is longer than 30 days, note this in your data processing records.

How do I archive data from vector databases?

Export the vectors and metadata for archived records to compressed files, then delete them from the live index. Store the archive files with the same naming convention as your document archives. If you need to restore archived vectors for an audit, re-insert them into a temporary collection. Keep the vector dimensionality and model version in the archive metadata so you know which embedding model produced them.

---

#DataRetention #GDPR #Compliance #DataLifecycle #Archival #AgenticAI #LearnAI #AIEngineering