Menguji sistem multiplayer AI agent yang beroperasi dalam lingkungan terdistribusi sering kali menghasilkan flaky test. Sumber kegagalan umumnya bukan logika inferensi model, melainkan mutasi state konkuren tak terduga, ketergantungan urutan resolusi event jaringan (network race conditions), dan penggunaan sleep() arbitrer dalam pengujian asinkron.

Untuk mengatasi masalah ini, implementasi pengujian harus beralih dari pendekatan real-time integration black-box ke deterministic test. Pola ini memisahkan aliran waktu fisik dari pemrosesan logika melalui event sourcing, clock virtual, dan snapshot state terisolasi.

Penyebab Utama Flaky State pada Multi-Agent

Dalam sistem di mana beberapa agen mengeksekusi rencana secara otonom dalam satu shared context, inkonsistensi state terjadi akibat tiga faktor teknis:

  • Non-deterministic Turn Resolution: Agen merespons event berbasis latensi I/O jaringan model LLM. Jika Agent A dan Agent B memicu mutasi state secara bersamaan, urutan commit ke state store menjadi tidak terprediksi.
  • Shared Mutable State: Penggunaan database transaksional tanpa versioning ketat memungkinkan operasi read-modify-write menghasilkan data stale saat dua agen memperbarui inventaris atau token workspace yang sama.
  • Asynchronous Polling Timeout: Penggunaan time.sleep() untuk menunggu respons agen memicu kegagalan intermiten ketika beban CPU runner CI/CD sedang tinggi.

Pola Solusi: Event Sourcing dan Virtual Clock

Menghilangkan flakiness mensyaratkan pemodelan state menggunakan Event Sourcing. Alih-alih menyimpan status akhir objek, simpan seluruh mutasi sebagai log event immutable yang terurut secara logis.

Pengujian deterministic dilakukan dengan mematikan clock sistem fisik (wall clock) dan menggantinya dengan Discrete Event Simulation (DES) atau Virtual Clock. Setiap langkah komputasi agen dan kedatangan pesan jaringan dipicu oleh ticks waktu logis, bukan milidetik fisik.

Implementasi Test Suite Deterministic

Berikut adalah implementasi Python yang mendemonstrasikan sistem multiplayer agent sederhana dengan event loop deterministik. Implementasi ini tidak menggunakan sleep() dan memvalidasi rekonsiliasi state via event log.

import heapq
from dataclasses import dataclass, field
from typing import Callable, List, Dict, Any

@dataclass(order=True)
class ScheduledEvent:
    timestamp: int
    event_id: int
    payload: Dict[str, Any] = field(compare=False)
    handler: Callable = field(compare=False)

class DeterministicVirtualClock:
    def __init__(self):
        self.current_time = 0
        self._counter = 0
        self._queue: List[ScheduledEvent] = []

    def schedule(self, delay: int, payload: Dict[str, Any], handler: Callable):
        self._counter += 1
        heapq.heappush(
            self._queue, 
            ScheduledEvent(self.current_time + delay, self._counter, payload, handler)
        )

    def tick(self) -> bool:
        if not self._queue:
            return False
        event = heapq.heappop(self._queue)
        self.current_time = event.timestamp
        event.handler(event.payload)
        return True

    def run_until_idle(self):
        while self.tick():
            pass

class AgentStateStore:
    def __init__(self):
        self.events: List[Dict[str, Any]] = []
        self.state: Dict[str, Any] = {"resources": 0, "locks": {}}

    def apply(self, event: Dict[str, Any]):
        self.events.append(event)
        event_type = event.get("type")
        
        if event_type == "CLAIM_RESOURCE":
            agent_id = event["agent_id"]
            amount = event["amount"]
            self.state["resources"] += amount
            self.state["locks"][agent_id] = self.state["locks"].get(agent_id, 0) + amount

class MockAIAgent:
    def __init__(self, agent_id: str, clock: DeterministicVirtualClock, store: AgentStateStore):
        self.agent_id = agent_id
        self.clock = clock
        self.store = store

    def receive_task(self, execution_delay: int, amount: int):
        # Menggantikan pemanggilan LLM nyata dengan simulasi waktu eksekusi diskrit
        self.clock.schedule(
            delay=execution_delay,
            payload={"type": "CLAIM_RESOURCE", "agent_id": self.agent_id, "amount": amount},
            handler=self.store.apply
        )

Test Suite Assertion Tanpa Polling

Dengan model di atas, pengujian dijalankan langsung hingga event log kosong. State diverifikasi secara deterministik pada titik waktu tertentu.

def test_multi_agent_deterministic_resource_allocation():
    clock = DeterministicVirtualClock()
    store = AgentStateStore()

    agent_alpha = MockAIAgent("agent_a", clock, store)
    agent_bravo = MockAIAgent("agent_b", clock, store)

    # Agent B lebih lambat (delay=50) dibanding Agent A (delay=20)
    agent_bravo.receive_task(execution_delay=50, amount=5)
    agent_alpha.receive_task(execution_delay=20, amount=10)

    # Eksekusi event pertama: Agent A harus commit pada tick 20
    clock.tick()
    assert clock.current_time == 20
    assert store.state["resources"] == 10
    assert "agent_a" in store.state["locks"]
    assert "agent_b" not in store.state["locks"]

    # Lanjutkan hingga semua agen selesai
    clock.run_until_idle()
    assert clock.current_time == 50
    assert store.state["resources"] == 15
    assert store.state["locks"]["agent_b"] == 5

    # Verifikasi integritas urutan event log
    assert len(store.events) == 2
    assert store.events[0]["agent_id"] == "agent_a"
    assert store.events[1]["agent_id"] == "agent_b"

if __name__ == "__main__":
    test_multi_agent_deterministic_resource_allocation()
    print("Test passed deterministically without race condition.")

Reproducible Context Mocking

Selain timing, variasi token dari respons LLM dapat memicu flakiness pada struktur payload state. Pisahkan prompt parser dan engine mutasi state:

  1. Record-Replay Strategy: Simpan hasil inferensi token LLM dalam bentuk fixture serial (JSON) yang terikat pada seed tertentu.
  2. Schema Validation: Jalankan mock agen yang mengembalikan struktur data tetap berbasis deterministic state machine daripada text completion bebas.
  3. Seed Isolation: Jika agen menggunakan library kalkulasi probabilitas untuk merencanakan langkah (misalnya Monte Carlo Tree Search atau random temperature), set nilai PRNG seed secara statis di tingkat runner test (contoh: random.seed(42)).

Trade-offs dan Kapan Menggunakannya

Peringatan Arsitektur: Deterministic testing berbasis Virtual Clock menguji orchestration logic, konsistensi data store, dan integrasi antar-agen. Pola ini tidak ditujukan untuk menguji kualitas output semantik atau akurasi penalaran teks model LLM itu sendiri.

Keuntungan dari pendekatan ini adalah kecepatan eksekusi tinggi (ratusan simulasi selesai dalam hitungan milidetik di CI/CD) dan reprodusibilitas 100% jika terjadi kegagalan. Batasannya adalah kebutuhan untuk membungkus semua operasi I/O dan library waktu ke dalam abstraksi clock virtual sistem.