JSONL Logging & Replay-based Testing

mock_openai.return_value = "fake response"처럼 모델 응답을 고정한 test는 계속 통과할 수 있어. 그 사이 SDK parameter, prompt, tool schema가 바뀌면 운영 환경만 깨져. 고정 mock은 실제 모델과 wire의 변화를 반영하지 못해.

실제 session을 capture해 replay해

실제 session event를 JSONL에 저장하고 CI에서 코드에 다시 흘려보내. 모델 event는 임의 fixture가 아니라 capture한 JSONL에서 읽어. final text, tool 순서, 전체 비용처럼 관찰 가능한 결과를 검증하면 logic 변화가 생긴 지점에서 실패해.

mock과 replay가 묻는 질문은 달라

mock — 'X로 호출하면 내가 정한 Y를 반환했을 때 코드가 동작하는가?'
replay — '실제 호출에서 받은 event sequence를 다시 넣으면 같은 final state에 도달하는가?'

cwkPippa JSONL은 저장 상태에서도 암호화돼

conversation별 JSONL line을 Fernet으로 암호화하고 passphrase는 office Mac Keychain에 둬. 다른 Mac에는 opaque blob으로 rsync되어 plaintext 파일로 남지 않아. 이 방식은 2026-04-28에 적용됐어.

Code

Append-only JSONL writer with line-level encryption·python

import json, time
from pathlib import Path
from datetime import datetime, timezone

class JSONLLogger:
    """Append-only JSONL logger for agent debugging."""
    def __init__(self, log_path):
        self.log_path = Path(log_path)
        self.log_path.parent.mkdir(parents=True, exist_ok=True)
        self._file = open(self.log_path, "a", buffering=1)  # line-buffered

    def log(self, event_type, **fields):
        record = {"ts": datetime.now(timezone.utc).isoformat(),
                  "event": event_type, **fields}
        self._file.write(json.dumps(record) + "\\n")

    def log_request(self, session_id, model, messages):
        start = time.monotonic()
        self.log("request", session_id=session_id, model=model,
                 message_count=len(messages))
        return start

    def log_response(self, session_id, start_time, finish_reason, usage, cost_usd):
        self.log("response", session_id=session_id,
                 latency_ms=round((time.monotonic() - start_time) * 1000),
                 finish_reason=finish_reason, cost_usd=round(cost_usd, 6))

Replay-based pytest fixture·python

class MockTransport(httpx.AsyncBaseTransport):
    """Return predetermined responses for deterministic testing."""
    def __init__(self, responses: dict):
        self._responses = responses  # url_path → response_body

    async def handle_async_request(self, request):
        body = self._responses.get(request.url.path, {"error": "Not found"})
        return httpx.Response(200, json=body, headers={"content-type": "application/json"})

# Use in tests:
mock = MockTransport({"/v1/chat/completions": {
    "choices": [{"message": {"content": "Test"}, "finish_reason": "stop"}],
    "usage": {"prompt_tokens": 10, "completion_tokens": 5, "total_tokens": 15},
}})
test_client = httpx.AsyncClient(base_url="https://api.openai.com", transport=mock)

실제 session을 capture해 replay해

mock과 replay가 묻는 질문은 달라

cwkPippa JSONL은 저장 상태에서도 암호화돼

Code

External links

Exercise

Progress

댓글 0