Ketika AI agent dijalankan pada arsitektur multi-worker untuk menangani tool calls asinkron secara paralel, akses konkuren ke buffer memori percakapan (persistent memory) menimbulkan race condition. Dua worker yang mengeksekusi tool berbeda secara simultan dapat membaca context snapshot yang sama, lalu melakukan penimpaan (clobbering) saat menyimpan ringkasan memori atau vector embeddings terbaru ke database. Artikel ini membahas penerapan state locking persistent memory agent guna memastikan konsistensi state percakapan dan mencegah hilangnya riwayat eksekusi.

Akar Masalah Concurrency pada Persistent Memory

Pola AI agent modern seperti Hermes atau ReAct memecah eksekusi kompleks ke dalam multiple tool calls. Dalam sistem backend terdistribusi, tool execution didelegasikan ke message queue (seperti Celery, BullMQ, atau RabbitMQ) dengan banyak worker aktif. Alur kerjanya rentan mengalami kegagalan konkurensi berikut:

  • Lost Update: Worker A dan Worker B membaca percakapan sesi session_123 pada versi v1. Keduanya menghasilkan inferensi terpisah. Worker A menulis v2, lalu Worker B menimpa v2 dengan outputnya, menghapus rekaman riwayat Worker A.
  • Vector Index Desynchronization: Embedding memori diperbarui menggunakan state yang usang, merusak representasi semantik retrieval augmented generation (RAG) agent pada konteks berikutnya.
  • Split-Brain Buffer: Terjadi percabangan dialog jika pesan pengguna baru disisipkan di antara siklus pembacaan dan penulisan memory checkpoint.

Strategi Penguncian: Redis Lock vs Database Row Lock

Dua pendekatan umum untuk serialisasi mutasi state adalah database row-level locking dan distributed lock via Redis.

1. Database Row-Level Locking (SELECT FOR UPDATE)

Pendekatan ini mengandalkan transaksi ACID relational database (PostgreSQL/MySQL). Ketika worker membaca sesi memori, ia mengunci baris tersebut secara eksklusif hingga transaksi selesai.

BEGIN;
SELECT session_id, memory_buffer, version 
FROM agent_sessions 
WHERE session_id = 'session_123' 
FOR UPDATE;

-- Jalankan pembaruan buffer memori
UPDATE agent_sessions 
SET memory_buffer = memory_buffer || '{\"role\": \"tool\", \"content\": \"result\"}'::jsonb, 
    version = version + 1 
WHERE session_id = 'session_123';
COMMIT;

Kelebihan: Konsistensi terjamin penuh tanpa ketergantungan infrastruktur tambahan.
Kekurangan: Menahan koneksi database selama worker menjalankan LLM inference atau external tool calls menyebabkan connection pool exhaustion.

2. Redis Distributed Lock (Redlock / Single Instance Lock)

Pola ini memisahkan layer sinkronisasi dari storage utama. Worker mengakuisisi lock di Redis dengan time-to-live (TTL) sebelum membaca memori dari storage, dan melepaskan lock setelah penulisan selesai.

Gunakan Redis lock untuk membatasi eksekusi I/O intensif, dan batasi SELECT FOR UPDATE hanya pada transaksi mutasi data yang cepat (milidetik).

Implementasi State Locking Persistent Memory Agent

Berikut implementasi Python minimalis menggunakan Redis distributed lock context manager untuk mutasi atomic state agent:

import time
import uuid
import redis

client = redis.Redis(host='localhost', port=6379, db=0)

class LockTimeoutError(Exception):
    pass

class AgentSessionLock:
    def __init__(self, session_id: str, timeout: float = 5.0, expire: int = 10):
        self.session_id = session_id
        self.lock_key = f"lock:agent:memory:{session_id}"
        self.timeout = timeout
        self.expire = expire
        self.identifier = str(uuid.uuid4())

    def __enter__(self):
        deadline = time.time() + self.timeout
        while time.time() < deadline:
            # Set NX EX memastikan atomisitas akuisisi kunci
            if client.set(self.lock_key, self.identifier, nx=True, ex=self.expire):
                return self
            time.sleep(0.05)
        raise LockTimeoutError(f"Timeout acquiring lock for session {self.session_id}")

    def __exit__(self, exc_type, exc_val, exc_tb):
        # Release lock menggunakan Lua script agar aman dari token mismatch
        release_script = """
        if redis.call("get", KEYS[1]) == ARGV[1] then
            return redis.call("del", KEYS[1])
        else
            return 0
        end
        """
        client.eval(release_script, 1, self.lock_key, self.identifier)

def append_agent_memory(session_id: str, new_event: dict):
    try:
        with AgentSessionLock(session_id, timeout=3.0, expire=10):
            # 1. Fetch persistent state
            raw_mem = client.get(f"memory:{session_id}")
            memory = raw_mem.decode('utf-8') if raw_mem else ""
            
            # 2. Mutasi state
            updated_memory = f"{memory}\n{new_event['role']}: {new_event['content']}".strip()
            
            # 3. Atomic write-back
            client.set(f"memory:{session_id}", updated_memory)
            
            # 4. Cache Invalidation / Flag refresh jika ada embedding layer
            client.publish(f"channel:memory_invalidated", session_id)
            
    except LockTimeoutError:
        # Fallback: Jadwalkan ulang tugas ke queue dengan exponential backoff
        reschedule_task(session_id, new_event)

def reschedule_task(session_id: str, event: dict):
    # Re-queue task dengan delay agar worker lain menyelesaikan eksekusi
    pass

Idempotency dan Invalidation Cache

State locking saja tidak cukup jika antrean worker mengalami retry akibat network partition. Terapkan prinsip berikut untuk menjaga integritas data:

  1. Task Idempotency Key: Setiap tool call harus membawa idempotency_key unik (misal kombinasi tool_call_id dan run_id). Simpan ID yang sudah diproses di hash table Redis. Jika ID ditemukan, lewati penulisan memori.
  2. Optimistic Concurrency Control (OCC) Fallback: Tambahkan kolom version numerik pada tabel memori database. Saat UPDATE dilakukan, sertakan predikat WHERE session_id = :id AND version = :current_version. Jika jumlah row terupdate 0, reload memori dan ulangi operasi.
  3. Cache Invalidation Pattern: Hindari cache drift dengan menerapkan invalidation ketat. Jangan langsung memperbarui Redis read-cache di dalam lock worker jika pipeline RAG membaca dari PostgreSQL. Hapus key cache agar pembacaan berikutnya memicu Cache-Aside fetch dari database primer.

Penanganan Lock Acquisition Timeout

Ketika worker gagal mendapatkan lock dalam batas waktu yang ditentukan, hindari blocking tak terbatas. Terapkan strategi mitigasi:

  • Exponential Backoff Re-queueing: Kembalikan job ke message queue dengan dead-letter exchange atau delay parameter (misal: 200ms, 400ms, 800ms).
  • Circuit Breaker: Jika lock tertahan melebihi TTL normal, monitor keberadaan worker zombie dan evaluasi apakah worker crash sebelum blok __exit__ sempat dipanggil.