Pengujian integrasi sistem multi-agent untuk rekayasa perangkat lunak sering menghasilkan flaky tests jika bergantung langsung pada keluaran model bahasa besar (LLM). Ketidakpastian teks keluaran (non-determinisme), latensi inferensi, dan biaya token membuat pengujian pipeline orkestrasi sulit direproduksi. Solusinya adalah memutus ketergantungan langsung ke model pada level integrasi harness dan memfokuskan validasi pada kontrak protokol pesan serta mutasi kode.

Arsitektur Determinisme Orkestrasi Multi-Agent

Verifikasi orkestrasi coding agent memerlukan isolasi total antara logika koordinasi dan mesin inferensi LLM. Sistem kolaborasi multi-agent bekerja sebagai state machine terdistribusi: Planner mendistribusikan subtask, Coder mengusulkan perubahan file, dan Reviewer memverifikasi patch. Pengujian harness harus menggantikan inferensi generatif dengan stubs berbasis protokol (mock protocol) yang menyimulasikan message passing melalui antarmuka asinkron seperti UNIX domain socket, pipes, atau in-memory queues.

Tiga komponen utama dalam harness deterministik:

  • Protocol Mock Engine: Mengirimkan respons pesan terstruktur (JSON-RPC atau custom schema) berdasarkan skenario pengujian tanpa pemanggilan API LLM.
  • IPC/Socket Message Broker: Menghubungkan agen melalui kanal komunikasi asinkron sebenarnya untuk memastikan serialization, deserialization, dan urutan pesan (sequencing) bekerja sesuai spesifikasi.
  • AST-Based Assertion: Mengabaikan perbedaan trivial seperti spasi, nama variabel dummy, atau komentar dengan membandingkan Abstract Syntax Tree (AST) sebelum dan sesudah patch diterapkan oleh coding agent.

Implementasi Test Harness dengan Pytest

Contoh berikut mengimplementasikan test harness minimal menggunakan pustaka standar Python (asyncio, ast) dan pytest. Skrip ini memverifikasi bahwa Orchestrator menerima instruksi, mengarahkan payload ke sub-agent, dan menghasilkan patch yang valid secara sintaksis.

import asyncio
import json
import ast
from typing import Dict, Any

class AgentProtocolMock:
    def __init__(self):
        self.canned_responses = {
            "planner": {"action": "delegate", "target": "coder", "task": "add_type_hints"},
            "coder": {"action": "patch", "file": "main.py", "patch": "def add(a: int, b: int) -> int:\n    return a + b"}
        }

    async def handle_message(self, message: Dict[str, Any]) -> Dict[str, Any]:
        role = message.get("role")
        if role in self.canned_responses:
            return self.canned_responses[role]
        raise ValueError(f"Unexpected role: {role}")

class MockOrchestrator:
    def __init__(self, protocol: AgentProtocolMock):
        self.protocol = protocol
        self.state = "IDLE"

    async def run_pipeline(self, initial_task: str) -> str:
        self.state = "PLANNING"
        plan = await self.protocol.handle_message({"role": "planner", "input": initial_task})
        
        if plan.get("action") == "delegate" and plan.get("target") == "coder":
            self.state = "CODING"
            result = await self.protocol.handle_message({"role": "coder", "input": plan["task"]})
            self.state = "COMPLETED"
            return result.get("patch", "")
        
        self.state = "FAILED"
        return ""

def assert_ast_equivalence(generated_code: str, expected_code: str) -> None:
    # Mengabaikan format spasi dan membandingkan struktur sintaksis
    tree_gen = ast.parse(generated_code)
    tree_exp = ast.parse(expected_code)
    assert ast.dump(tree_gen) == ast.dump(tree_exp), "AST patch tidak cocok dengan baseline expected"

Pengujian Terisolasi Tanpa Dependensi LLM

Kode uji berikut membuktikan orkestrasi selesai dengan benar dan patch kode yang dikeluarkan valid secara semantik sintaksis:

import pytest

@pytest.mark.asyncio
async def test_orchestration_coder_pipeline():
    mock_engine = AgentProtocolMock()
    orchestrator = MockOrchestrator(protocol=mock_engine)
    
    initial_prompt = "Tambahkan type hints pada fungsi add"
    final_patch = await orchestrator.run_pipeline(initial_prompt)
    
    assert orchestrator.state == "COMPLETED"
    
    # Verifikasi dengan AST diff
    expected_patch = "def add(a: int, b: int) -> int:\n    return a + b\n"
    assert_ast_equivalence(final_patch, expected_patch)

Penggunaan ast.dump() memastikan pengujian tidak gagal hanya karena perbedaan trailing whitespace, single/double quote, atau baris kosong yang kerap terjadi pada representasi patch.

Checklist Pencegahan Regresi State Machine

Ketika banyak agen saling bertukar context dan mengeksekusi patch bersamaan, risiko race condition atau transisi state yang tidak valid meningkat. Gunakan checklist berikut untuk audit skenario pengujian:

  1. Strict Transition Mapping: Pastikan state machine menolak transisi ilegal. Misalnya, Reviewer tidak boleh mengirim payload langsung ke Merge jika status masih CHANGES_REQUESTED.
  2. Deadlock & Circular Message Guard: Buat timeout tegas pada layer mock socket jika Agent A dan Agent B terjebak dalam loop klarifikasi context tanpa progres commit file.
  3. Context Truncation Handlers: Uji perilaku orkestrator saat ukuran pesan melampaui buffer IPC atau payload message format terpotong (malformed JSON).
  4. Patch Reentrancy Validation: Pastikan jika pipeline gagal di tengah eksekusi, working tree repository dikembalikan ke hash Git asal (idempotent rollback).