Agent State는 어디에 있어야 하는가: PostgreSQL, Redis, Checkpoint

이전 글 Worker는 Thread가 아니다에서는 장시간 Agent run의 수명 주기를 API 요청에서 queue와 worker로 넘겼다. 이제 worker가 실행을 맡았다고 해보자.

Worker는 문서 수집과 정규화를 끝내고 규칙 적용 단계로 넘어갔다. 진행률은 65%다. 사용자는 브라우저를 새로고침했고, 그 순간 worker Pod가 재시작됐다.

여기서 “state”라는 한 단어가 서로 다른 질문을 가리킨다.

• 이 run은 누구의 요청이며 최종적으로 성공했는가?
• 지금 어떤 단계를 실행 중이고 진행률은 몇 퍼센트인가?
• Worker가 다시 시작하면 어느 node부터 재개해야 하는가?
• 현재 함수가 쓰는 HTTP client와 임시 파싱 버퍼는 어디에 있는가?

이 네 상태는 수명도, 읽는 주체도, 실패 대응 요구사항도 다르다. 전부 Python dictionary 하나에 넣으면 프로세스 재시작을 견디지 못한다. 전부 PostgreSQL row 하나에 넣으면 작은 진행률 갱신마다 영속적 row를 계속 갱신하게 된다. 전부 Redis에 넣으면 빠르지만 업무 기록과 재개 의미 체계가 캐시 정책에 묶인다.

Agent state 설계는 데이터베이스 선택 문제가 아니라 서로 다른 종류의 진실과 복구 계약을 분리하는 문제다.

먼저 논리적 state와 물리적 저장소를 나눈다

논리적 artifact	답해야 하는 질문	필요한 수명	자연스러운 위치
Durable run record	누가 무엇을 요청했고 어떻게 끝났는가?	서비스·프로세스 재시작 이후에도 보존	PostgreSQL
Hot projection	지금 어느 단계이며 heartbeat는 최신인가?	실행 중 빠르게 읽고, 이후 만료 가능	Redis
Execution checkpoint	어디에서 안전하게 다시 시작하는가?	Run의 재개·보존 정책만큼	PostgreSQL 등 내구성 저장소
Process-local state	현재 함수에만 필요한가?	프로세스 수명 이하	Worker memory

먼저 묻는다이 값은 어떤 실패를 견뎌야 하는가?

1. 01
   Durable run record소유권 · status · 결과 · 감사
   →PostgreSQL
2. 02
   Hot projection현재 단계 · 진행률 · heartbeat
   →Redis
3. 03
   Execution checkpointvalues · 다음 단계 · 완료 작업
   →PostgreSQL 등 내구성 저장소
4. 04
   Process-local stateclient · buffer · coroutine
   →Worker memory

Checkpoint는 제품명이 아니라 재개 가능한 snapshot이라는 의미다.

Checkpoint는 데이터베이스 제품명이 아니다. 재개 가능한 snapshot이라는 데이터의 의미다. PostgreSQL에도 저장할 수 있고 요구하는 영속성에 따라 다른 backend를 사용할 수도 있다. 반대로 Redis에 값을 저장했다고 모두 캐시인 것도 아니다. 제품 이름이 역할을 자동으로 결정하지 않는다.

Redis 전체를 잃어도 run의 존재와 최종 결과는 설명할 수 있어야 한다. Worker 메모리 전체를 잃어도 checkpoint 이후부터는 다시 실행할 수 있어야 한다.

저장 위치는 일곱 질문으로 결정한다

저장소 이름보다 먼저 다음을 묻는다.

1. 프로세스가 죽어도 남아야 하는가?
2. 사용자가 며칠 뒤에도 조회하거나 감사해야 하는가?
3. 초당 여러 번 바뀌고 여러 client가 빠르게 읽는가?
4. 이 값만으로 worker가 안전하게 재개할 수 있는가?
5. TTL이나 eviction으로 사라져도 다시 만들 수 있는가?
6. 동시에 쓰는 주체가 둘 이상인가?
7. 배포 뒤에도 이전 run을 읽어야 하는가?

요구사항	설계 결과
감사·권한·최종 결과의 기준	관계와 transaction을 가진 durable record
빈번한 progress 조회, 재구성 가능	짧은 TTL의 hot projection
Worker crash 뒤 재개	직렬화 가능하고 versioned된 checkpoint
현재 함수 안에서만 사용	process-local object
여러 writer 또는 재시도	version, attempt identity, fencing
장기 변경 이력	별도 append-only event 또는 audit store

이 글은 초기·중간 규모의 내부 Agent 서비스를 기준으로 PostgreSQL과 Redis를 사용한다. 다른 트래픽·규정·팀 역량이면 다른 제품이 맞을 수 있다. 중요한 것은 제품 조합이 아니라 위 계약을 지키는 것이다.

하나의 run은 세 개의 저장 표현을 만든다

문서 검토 Agent가 다음 순서로 실행된다고 하자.

collect → normalize → apply_rules → draft → approval

같은 시점의 state를 저장소별로 펼치면 다음과 같다.

PostgreSQL: agent_runs
  run_id          = 7d5...
  tenant_id       = acme
  session_id      = 91a...
  status          = running
  input           = {"document_set": "travel-policy-2026"}
  final_result    = null
  version         = 3

Redis: agent:run:{7d5...}:state
  status          = running
  current_step    = apply_rules
  progress        = 65
  attempt_id      = 4fe...
  heartbeat_at    = 2026-10-04T04:10:00Z
  version         = 18

Checkpoint 12
  values          = {document refs, normalized facts, applicable rules}
  next_nodes      = [apply_rules]
  completed_nodes = [collect, normalize]

이 셋은 같은 state의 복제본이 아니다. 서로 다른 독자와 복구 계약을 가진 표현이다.

• Durable run record는 “run은 존재하며 아직 종료되지 않았다”를 보존한다.
• Hot projection은 UI와 운영자가 지금 보고 싶은 최신 값만 제공한다.
• Execution checkpoint는 새 worker가 계산을 이어갈 충분한 snapshot을 보존한다.

Durable run record: PostgreSQL 기준 정보를 작게 유지한다

Run table은 공식 기록이다. 사용자에게 결과를 보여주고, 권한을 검사하고, 과금이나 감사의 근거로 사용할 수 있어야 한다.

먼저 자주 검색하고 조인하는 값은 관계형 column으로 둔다.

CREATE TABLE agent_runs (
    run_id UUID PRIMARY KEY,
    tenant_id UUID NOT NULL,
    user_id UUID NOT NULL,
    session_id UUID NOT NULL,

    status TEXT NOT NULL CHECK (
        status IN ('queued', 'running', 'waiting_approval',
                   'succeeded', 'failed', 'cancelled')
    ),
    current_attempt_id UUID,
    fencing_token BIGINT NOT NULL DEFAULT 0,

    input JSONB NOT NULL,
    final_result JSONB,
    error JSONB,

    version BIGINT NOT NULL DEFAULT 0,
    created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
    started_at TIMESTAMPTZ,
    finished_at TIMESTAMPTZ,
    updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
);

CREATE INDEX agent_runs_session_created_idx
    ON agent_runs (tenant_id, session_id, created_at DESC);

CREATE INDEX agent_runs_active_idx
    ON agent_runs (tenant_id, status, updated_at)
    WHERE status IN ('queued', 'running', 'waiting_approval');

tenant_id, session_id, status, timestamp를 JSONB 안에 감추지 않은 이유가 있다.

• 권한 확인 쿼리와 foreign key를 명확하게 만들 수 있다.
• 활성 run이나 session 이력을 일반 B-tree index로 찾을 수 있다.
• Status에 CHECK constraint를 걸어 잘못된 전환의 일부를 막을 수 있다.
• 운영 쿼리가 매번 JSON path를 해석하지 않아도 된다.

current_attempt_id와 fencing_token은 queue claim을 획득할 때 transaction으로 갱신한다. 이후 worker의 쓰기는 이 권한 세대와 일치할 때만 받아들인다.

반면 제공자별 모델 옵션, 도구별 결과, 중첩된 가정처럼 구조가 유연한 값은 JSONB에 잘 맞는다. PostgreSQL의 jsonb는 JSON을 분해된 이진 형식으로 저장해 처리할 때 다시 파싱할 필요가 없고 색인도 지원한다. 다만 문서 구조가 유연하더라도 어느 정도 예측 가능한 schema를 유지하라고 공식 문서도 권장한다. (PostgreSQL JSON Types)

JSONB는 schema를 없애는 도구가 아니다

좋은 구분은 다음과 같다.

값	권장 형태	이유
tenant_id, session_id	관계형 column	조인·권한 확인·색인
status, created_at	관계형 column	자주 필터·정렬
LLM 제공자 옵션	JSONB	제공자마다 선택 field가 다름
도구 출력 payload	JSONB 또는 별도 table	구조가 다양하지만 감사 범위 고려
Token 사용량 합계	관계형 column	집계와 과금 쿼리가 잦다면 column
전체 메시지 이력	별도 rows	한 JSONB 문서의 무한 성장을 피함

큰 state 문서 하나를 매 단계마다 통째로 갱신하는 것은 피해야 한다. PostgreSQL은 JSONB column 일부만 바꾸는 표현을 지원하지만, 동시성 제어 관점에서는 row 갱신이다. 공식 문서도 큰 JSON 문서 갱신이 row 단위 잠금을 획득하므로 문서 크기와 잠금 경합을 고려하라고 설명한다.

즉 다음 구조는 편해 보여도 시간이 지나면 부담이 된다.

-- 모든 대화, event, tool result, checkpoint를 한 row에 계속 append
UPDATE agent_runs
SET state = state || $1::jsonb
WHERE run_id = $2;

Run 기록, 메시지, 도구 실행, checkpoint처럼 수명과 쿼리 패턴이 다른 데이터는 별도 table로 나누는 편이 낫다.

갱신에는 version과 허용된 전환을 둔다

Worker, 취소 API, 승인 처리기가 같은 run을 동시에 갱신할 수 있다. 마지막 쓰기가 무조건 이기는 방식이면 늦게 도착한 running 갱신이 이미 기록된 cancelled를 덮을 수 있다.

version을 이용한 낙관적 동시성 제어는 오래된 값을 쓰려는 주체를 감지하는 단순한 방법이다.

import json
from dataclasses import dataclass
from typing import Any

@dataclass(frozen=True)
class ConcurrentRunUpdate(Exception):
    run_id: str
    expected_version: int

async def transition_run(
    conn,
    *,
    run_id: str,
    expected_version: int,
    next_status: str,
    final_result: dict[str, Any] | None = None,
) -> int:
    row = await conn.fetchrow(
        """
        UPDATE agent_runs
        SET status = $1,
            final_result = COALESCE($2::jsonb, final_result),
            version = version + 1,
            updated_at = NOW(),
            finished_at = CASE
                WHEN $1 IN ('succeeded', 'failed', 'cancelled') THEN NOW()
                ELSE finished_at
            END
        WHERE run_id = $3
          AND version = $4
        RETURNING version
        """,
        next_status,
        json.dumps(final_result) if final_result is not None else None,
        run_id,
        expected_version,
    )

    if row is None:
        raise ConcurrentRunUpdate(run_id, expected_version)

    return row["version"]

Version 불일치가 발생하면 무조건 같은 갱신을 재시도해서는 안 된다. 최신 row를 다시 읽고, cancelled → running 같은 금지된 전환인지 판단해야 한다. State machine 규칙은 애플리케이션 코드나 데이터베이스 constraint/trigger로 명시적으로 강제한다.

ALLOWED_TRANSITIONS = {
    "queued": {"running", "cancelled"},
    "running": {"waiting_approval", "succeeded", "failed", "cancelled"},
    "waiting_approval": {"running", "cancelled"},
    "succeeded": set(),
    "failed": set(),
    "cancelled": set(),
}

def assert_transition(current: str, target: str) -> None:
    if target not in ALLOWED_TRANSITIONS[current]:
        raise ValueError(f"invalid run transition: {current} -> {target}")

LLM이 status를 결정하게 두지 않는 것도 중요하다. 모델은 “분석이 끝났다”고 말할 수 있지만 종료 전환은 필수 출력 검증과 영속화가 완료된 뒤 결정론적 코드가 수행해야 한다.

Hot projection: Redis에는 작고 현재적인 값만 둔다

Redis에는 UI와 운영 loop가 자주 읽는 작은 현재 상태 투영값을 둔다.

Key
agent:run:{run_id}:state

Hash fields
status          running
current_step    apply_rules
progress        65
attempt_id      4fe...
heartbeat_at    2026-10-04T04:10:00Z
updated_at      2026-10-04T04:10:00Z
version         18
detail_json     {"policy":"travel-2026","rules_checked":4}

Redis Hash는 field-value 쌍의 모음이다. HSET으로 일부 field를 갱신하고 HMGET이나 HGETALL로 읽을 수 있어 평평한 현재 상태 객체에 잘 맞는다. Hash field 값은 기본적으로 string/bytes이므로 중첩 객체는 JSON string으로 직렬화하거나 Redis JSON 같은 별도 자료형을 선택해야 한다. (Redis Hashes)

여기서는 별도 module 의존 없이 Hash를 사용한다.

import json
from datetime import UTC, datetime
from redis.asyncio import Redis

HOT_STATE_TTL_SECONDS = 24 * 60 * 60

def run_state_key(run_id: str) -> str:
    # 중괄호는 Redis Cluster에서 같은 run의 관련 key를
    # 같은 hash slot에 묶고 싶을 때 사용할 수 있다.
    return f"agent:run:{{{run_id}}}:state"

async def initialize_hot_state(
    redis: Redis,
    *,
    run_id: str,
    status: str,
    current_step: str,
    progress: int,
    attempt_id: str,
    fencing_token: int,
    version: int,
    detail: dict,
) -> None:
    if not 0 <= progress <= 100:
        raise ValueError("progress must be between 0 and 100")

    now = datetime.now(UTC).isoformat()
    key = run_state_key(run_id)

    async with redis.pipeline(transaction=True) as pipe:
        pipe.hset(
            key,
            mapping={
                "status": status,
                "current_step": current_step,
                "progress": progress,
                "attempt_id": attempt_id,
                "fencing_token": fencing_token,
                "heartbeat_at": now,
                "updated_at": now,
                "version": version,
                "detail_json": json.dumps(detail),
            },
        )
        pipe.expire(key, HOT_STATE_TTL_SECONDS)
        await pipe.execute()

이 초기화 함수는 queue claim을 얻은 현재 attempt만 호출한다. HSET과 EXPIRE를 transaction pipeline에 넣은 이유는 둘을 하나의 단위로 실행하기 위해서다. State 쓰기만 성공하고 프로세스가 죽어 TTL 설정이 빠지면 임시 key가 영구히 남을 수 있다. Redis transaction은 MULTI/EXEC 사이의 명령을 하나의 격리된 작업으로 실행하고, redis-py pipeline은 명령 묶음 처리와 transaction을 지원한다. 이후 갱신은 아래의 fencing 검사를 거친다. (Redis Transactions, redis-py Pipelines)

읽을 때는 저장 표현을 도메인 자료형으로 복원한다.

from dataclasses import dataclass

@dataclass(frozen=True)
class HotRunState:
    status: str
    current_step: str
    progress: int
    attempt_id: str
    fencing_token: int
    updated_at: str
    version: int
    detail: dict

async def read_hot_state(
    redis: Redis,
    run_id: str,
) -> HotRunState | None:
    raw = await redis.hgetall(run_state_key(run_id))
    if not raw:
        return None

    return HotRunState(
        status=raw["status"],
        current_step=raw["current_step"],
        progress=int(raw["progress"]),
        attempt_id=raw["attempt_id"],
        fencing_token=int(raw["fencing_token"]),
        updated_at=raw["updated_at"],
        version=int(raw["version"]),
        detail=json.loads(raw.get("detail_json", "{}")),
    )

이 예시는 Redis client를 decode_responses=True로 생성했다고 가정한다. 그렇지 않으면 key와 값이 bytes로 반환되므로 디코딩 계층이 필요하다.

Key에는 식별자와 수명을 함께 설계한다

Key 이름은 단순한 미관이 아니다. 누가 읽을 수 있고, 무엇과 함께 삭제되며, Cluster에서 어떤 다중 key 작업이 가능한지에 영향을 준다.

권장 시작점은 다음과 같다.

agent:run:{run_id}:state
agent:run:{run_id}:lease
agent:run:{run_id}:cancel

여기서 {run_id}는 Redis Cluster hash tag다. 같은 중괄호 값을 가진 key는 같은 hash slot으로 배치할 수 있다. 다중 key transaction이나 Lua script가 필요한 관련 key를 같은 slot에 두는 데 유용하다. 하지만 모든 tenant key를 {tenant_id} 하나로 묶으면 큰 tenant가 한 slot에 몰릴 수 있다. 실제 접근 패턴을 보고 hash tag를 정해야 한다.

더 중요한 보안 원칙이 있다.

Redis key에 tenant id가 들어 있다고 권한 확인이 되는 것은 아니다.

API는 PostgreSQL의 tenant_id와 인증된 주체를 기준으로 run 접근 권한을 확인한 뒤 Redis를 읽어야 한다. 사용자가 전달한 run_id로 Redis를 바로 조회하고 결과를 반환하면 tenant 간 데이터 유출이 생길 수 있다.

TTL은 정리 규칙이자 의미다

실시간 state는 영원히 남을 필요가 없다. 실행 중에는 갱신할 때마다 TTL을 연장하고, 종료 상태가 PostgreSQL에 기록된 뒤에는 더 짧은 TTL로 바꿀 수 있다.

RUNNING_TTL = 24 * 60 * 60
TERMINAL_TTL = 60 * 60

async def mark_hot_state_terminal(
    redis: Redis,
    *,
    run_id: str,
    status: str,
    version: int,
) -> None:
    if status not in {"succeeded", "failed", "cancelled"}:
        raise ValueError("terminal status required")

    key = run_state_key(run_id)
    async with redis.pipeline(transaction=True) as pipe:
        pipe.hset(
            key,
            mapping={
                "status": status,
                "progress": 100 if status == "succeeded" else 0,
                "version": version,
                "updated_at": datetime.now(UTC).isoformat(),
            },
        )
        pipe.expire(key, TERMINAL_TTL)
        await pipe.execute()

Redis의 EXPIRE는 key 전체에 TTL을 설정한다. HSET처럼 값을 제자리에서 변경하는 명령은 기존 key 만료 시간을 제거하지 않지만, SET처럼 key 값을 교체하는 작업은 만료 시간을 지울 수 있다. 구현을 바꿀 때 TTL이 계속 유지된다고 가정하지 말고 명령의 의미를 확인해야 한다. (EXPIRE)

TTL이 만료되었다고 run이 사라진 것은 아니다. Redis에서 찾지 못하면 API는 PostgreSQL의 영속적 state를 대신 사용해야 한다. 이것이 Redis를 투영값으로 두는 장점이다.

GET /runs/{run_id}
   │
   ├─ PostgreSQL: authorization + durable status/result
   │
   └─ Redis: current step/progress가 있으면 응답에 보강

갱신 순서에는 attempt와 fencing이 필요하다

병렬 node나 재시도가 있는 Agent에서는 늦게 도착한 갱신이 최신 진행률을 덮을 수 있다.

Worker A writes version 19, progress 70
Worker B arrives late with version 18, progress 60

Blind HSET result → UI가 70%에서 60%로 되돌아감

한 run의 조정자만 진행률을 쓰도록 책임을 좁히는 것이 가장 단순하다. 그래도 오래된 갱신 가능성이 있다면 version 비교와 쓰기를 Redis 안에서 원자적으로 실행해야 한다. 읽은 후 애플리케이션에서 비교하고 다시 HSET하면 두 명령 사이에 경쟁 상태가 생긴다.

Lua script는 비교와 갱신을 한 서버 측 작업으로 묶을 수 있다.

UPDATE_IF_NEWER = """
local current_fence = tonumber(redis.call('HGET', KEYS[1], 'fencing_token'))
local current_version = tonumber(redis.call('HGET', KEYS[1], 'version'))
local proposed_fence = tonumber(ARGV[1])
local proposed_version = tonumber(ARGV[2])

if current_fence and proposed_fence < current_fence then
    return 0
end

if current_fence == proposed_fence
   and current_version
   and proposed_version <= current_version then
    return 0
end

redis.call(
    'HSET', KEYS[1],
    'fencing_token', ARGV[1],
    'version', ARGV[2],
    'attempt_id', ARGV[3],
    'status', ARGV[4],
    'current_step', ARGV[5],
    'progress', ARGV[6],
    'updated_at', ARGV[7]
)
redis.call('EXPIRE', KEYS[1], ARGV[8])
return 1
"""

async def update_if_newer(
    redis: Redis,
    *,
    run_id: str,
    attempt_id: str,
    fencing_token: int,
    version: int,
    status: str,
    current_step: str,
    progress: int,
) -> bool:
    result = await redis.eval(
        UPDATE_IF_NEWER,
        1,
        run_state_key(run_id),
        fencing_token,
        version,
        attempt_id,
        status,
        current_step,
        progress,
        datetime.now(UTC).isoformat(),
        RUNNING_TTL,
    )
    return result == 1

Redis transaction의 WATCH/MULTI/EXEC로 낙관적 잠금을 구현할 수도 있다. WATCH한 key가 EXEC 전에 바뀌면 transaction이 중단되고 client가 최신 값을 읽어 다시 판단한다. Lua와 WATCH 중 무엇을 쓸지는 갱신 복잡도와 client 재시도 요구에 따라 고른다.

version은 같은 attempt 안의 순서를, fencing_token은 attempt 사이의 쓰기 권한을 구분한다. 새 queue claim이 더 큰 token을 받으면 이전 worker가 더 큰 local version을 만들더라도 갱신이 거부된다. attempt_id는 운영자가 어떤 실행 시도가 값을 썼는지 추적하는 식별자다.

Redis Hash에 넣지 않을 것을 정한다

Hash가 편하다고 모든 state를 field로 계속 추가하면 또 하나의 거대한 문서가 된다.

다음은 피하는 편이 좋다.

• 전체 prompt와 모든 LLM response
• 수천 개의 event를 event_1, event_2 field로 누적
• 큰 문서 모음과 이진 파일
• 유일한 final result 복사본
• 감사에 필요한 도구 실행 이력
• 재개에 필요한 유일한 checkpoint

Redis Hash는 작고 현재적인 투영값에 적합하다. 중첩 state가 크고 부분 쿼리가 필요하면 Redis JSON을 검토할 수 있지만, 데이터 모델의 의미가 달라지는 것은 아니다. Event 이력에는 append 의미가 필요하고, snapshot에는 snapshot 의미가 필요하다. 7편에서 Hash state와 Redis Stream event를 구분한다.

HGETALL은 field 수에 비례해 응답이 커진다. 작은 state object에는 괜찮지만 hash가 커지면 필요한 field만 HMGET으로 읽는 편이 낫다. Redis 공식 문서도 HGETALL을 hash 크기 N에 대한 O(N) command로 분류한다.

Redis에 RDB나 AOF 영속성을 켜도 역할이 자동으로 바뀌지는 않는다. (Redis Persistence) 기준 정보는 디스크 쓰기 여부뿐 아니라 transaction, constraint, backup, 장기 query, 감사와 failover의 허용 손실 범위로 결정된다. Eviction이 켜져 있으면 TTL 전에도 key가 사라질 수 있으므로 hot projection은 여전히 재구성 가능해야 한다. (Key eviction)

Execution checkpoint: 현재 status가 아니라 재개 snapshot이다

Redis hot state가 다음과 같다고 해보자.

{
  "status": "running",
  "current_step": "apply_rules",
  "progress": 65
}

이 정보만으로 apply_rules를 다시 시작할 수는 없다. 어떤 문서와 정규화된 사실을 사용했고, 어떤 규칙이 적용 대상이며, 어느 dependency가 끝났는지 알 수 없기 때문이다.

Checkpoint는 실행을 재개할 수 있는 snapshot이다.

from dataclasses import dataclass, field
from typing import Any

@dataclass(frozen=True)
class AgentCheckpoint:
    run_id: str
    checkpoint_id: str
    sequence: int
    values: dict[str, Any]
    next_nodes: list[str]
    completed_nodes: list[str]
    metadata: dict[str, Any] = field(default_factory=dict)

Checkpoint에는 보통 다음 정보가 필요하다.

• Graph state 값
• 다음에 실행할 node 또는 준비된 집합
• 완료된 node와 그 출력 참조
• 대기 중인 승인이나 중단 정보
• Checkpoint sequence와 parent checkpoint
• State schema version
• 재현에 필요한 모델/도구 설정 참조

반대로 열린 데이터베이스 연결, HTTP client, coroutine 객체처럼 직렬화할 수 없고 프로세스에 종속된 객체는 넣지 않는다. 새 worker가 자원을 다시 만들고, checkpoint에는 자원을 재구성할 식별자와 데이터만 남긴다.

영속적 checkpoint를 PostgreSQL에 저장하는 예

Framework에 독립적인 schema는 다음처럼 시작할 수 있다.

CREATE TABLE agent_checkpoints (
    run_id UUID NOT NULL REFERENCES agent_runs(run_id),
    checkpoint_id UUID NOT NULL,
    sequence BIGINT NOT NULL,
    parent_checkpoint_id UUID,

    state_schema_version INTEGER NOT NULL,
    values JSONB NOT NULL,
    next_nodes TEXT[] NOT NULL,
    completed_nodes TEXT[] NOT NULL,
    metadata JSONB NOT NULL DEFAULT '{}',

    created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),

    PRIMARY KEY (run_id, checkpoint_id),
    UNIQUE (run_id, sequence)
);

CREATE INDEX agent_checkpoints_latest_idx
    ON agent_checkpoints (run_id, sequence DESC);

저장과 조회 코드는 추가 전용 snapshot처럼 다룬다.

import json

async def save_checkpoint(conn, checkpoint: AgentCheckpoint) -> None:
    await conn.execute(
        """
        INSERT INTO agent_checkpoints (
            run_id,
            checkpoint_id,
            sequence,
            parent_checkpoint_id,
            state_schema_version,
            values,
            next_nodes,
            completed_nodes,
            metadata
        ) VALUES ($1, $2, $3, $4, $5, $6::jsonb, $7, $8, $9::jsonb)
        """,
        checkpoint.run_id,
        checkpoint.checkpoint_id,
        checkpoint.sequence,
        checkpoint.metadata.get("parent_checkpoint_id"),
        checkpoint.metadata.get("state_schema_version", 1),
        json.dumps(checkpoint.values),
        checkpoint.next_nodes,
        checkpoint.completed_nodes,
        json.dumps(checkpoint.metadata),
    )

async def load_latest_checkpoint(conn, run_id: str):
    return await conn.fetchrow(
        """
        SELECT *
        FROM agent_checkpoints
        WHERE run_id = $1
        ORDER BY sequence DESC
        LIMIT 1
        """,
        run_id,
    )

Checkpoint를 같은 row에 계속 덮어쓰지 않고 sequence별로 남기면 어느 snapshot까지 영속화했는지 알 수 있고, 디버깅이나 human-in-the-loop 재개에도 도움이 된다. 대신 저장 공간이 무한히 증가하지 않도록 보존과 압축 정책이 필요하다.

모든 node 뒤에 전체 state를 저장해야 하는 것도 아니다. Checkpoint frequency에는 trade-off가 있다.

자주 저장	드물게 저장
실패 후 재실행 범위가 작다	쓰기와 저장 비용이 작다
DB 쓰기와 직렬화 비용이 커진다	비싼 도구 호출을 다시 수행할 수 있다
HITL 중단 위치가 세밀하다	재개 지점이 거칠다

부수 효과 직전·직후, 비싼 도구 호출 완료 후, HITL 중단 전에 checkpoint를 두는 식으로 복구 비용을 기준으로 결정한다.

저장 빈도와 schema version을 함께 설계한다

오늘 state가 다음 모양이라고 하자.

{
  "policy": {"id": "travel-2026", "revision": 3}
}

다음 배포에서 단일 policy를 policy_refs 배열로 바꾸면 이전 checkpoint를 새 코드가 읽지 못할 수 있다. 따라서 checkpoint에는 state_schema_version을 두고 마이그레이션 또는 호환 계층을 마련해야 한다.

def migrate_checkpoint(values: dict, from_version: int) -> dict:
    if from_version == 1:
        policy = values.pop("policy", None)
        if policy is not None:
            values["policy_refs"] = [policy]
        from_version = 2

    if from_version != 2:
        raise ValueError(f"unsupported checkpoint version: {from_version}")

    return values

장시간 run이 여러 배포 version을 가로지를 수 있다면 코드 version도 metadata에 남기고, 새 worker가 이전 checkpoint를 읽을 수 있는 기간을 출시 정책에 포함해야 한다.

Worker code에서 세 저장소를 연결해 보자

이제 run 하나가 실제로 움직이는 순서를 합쳐보자.

async def execute_run(
    *,
    run_id: str,
    attempt_id: str,
    fencing_token: int,
    postgres,
    redis: Redis,
) -> None:
    async with postgres.acquire() as conn:
        run = await conn.fetchrow(
            "SELECT * FROM agent_runs WHERE run_id = $1",
            run_id,
        )
        checkpoint = await load_latest_checkpoint(conn, run_id)

    state = restore_state(run, checkpoint)
    hot_version = 0

    for step in ready_steps(state):
        hot_version += 1
        await update_if_newer(
            redis,
            run_id=run_id,
            attempt_id=attempt_id,
            fencing_token=fencing_token,
            version=hot_version,
            status="running",
            current_step=step.name,
            progress=estimate_progress(state, step),
        )

        output = await step.execute(state)
        state = apply_output(state, step, output)

        if should_checkpoint(step, state):
            async with postgres.acquire() as conn:
                await save_checkpoint(
                    conn,
                    build_checkpoint(run_id, state),
                )

    result = build_final_result(state)

    async with postgres.acquire() as conn:
        current = await conn.fetchrow(
            "SELECT status, version FROM agent_runs WHERE run_id = $1",
            run_id,
        )
        assert_transition(current["status"], "succeeded")
        durable_version = await transition_run(
            conn,
            run_id=run_id,
            expected_version=current["version"],
            next_status="succeeded",
            final_result=result,
        )

    # Durable terminal result가 먼저다. Redis는 그다음 최신 projection을 맞춘다.
    await mark_hot_state_terminal(
        redis,
        run_id=run_id,
        status="succeeded",
        version=durable_version,
    )

이 코드는 중요한 쓰기 순서를 보여준다.

1. 실행 중 진행률은 Redis에 자주 쓴다.
2. 재개 지점에서는 영속적 checkpoint를 저장한다.
3. 최종 결과와 종료 status는 PostgreSQL에 먼저 commit한다.
4. 그 후 Redis 투영값을 종료 상태로 맞춘다.

마지막 Redis 쓰기가 실패해도 PostgreSQL에는 성공 결과가 남는다. API가 영속적 row를 우선하면 사용자에게 잘못된 running 상태를 영원히 보여주지 않는다. 조정 작업이 종료 run의 Redis 투영값을 정리할 수도 있다.

쓰기와 조회 경로는 durable state를 우선한다

쓰기 경로

1. 1Redis실행 중 progress
2. 2Checkpoint store안전한 재개 지점
3. 3PostgreSQL최종 결과 commit
4. 4Redis종료 projection 정리

최종 결과 commit 전에는 성공으로 ack하지 않는다.

조회 경로

1. 1PostgreSQL권한·durable status 확인
2. 2Redis활성 run만 progress 보강
3. 3APIterminal status는 DB 우선

Redis timeout이나 miss는 제한 모드이지 run 소실이 아니다.

API는 영속적 state와 hot projection을 합쳐 읽는다

Status endpoint는 Redis만 읽으면 안 된다. 먼저 tenant 권한과 영속적 state를 확인하고, run이 활성 상태일 때 실시간 투영값을 보강한다.

from fastapi import HTTPException

async def get_run_view(
    *,
    run_id: str,
    tenant_id: str,
    conn,
    redis: Redis,
) -> dict:
    run = await conn.fetchrow(
        """
        SELECT run_id, tenant_id, status, current_attempt_id, fencing_token,
               final_result, error,
               created_at, updated_at
        FROM agent_runs
        WHERE run_id = $1 AND tenant_id = $2
        """,
        run_id,
        tenant_id,
    )
    if run is None:
        raise HTTPException(404, "run not found")

    response = {
        "run_id": str(run["run_id"]),
        "status": run["status"],
        "result": run["final_result"],
        "error": run["error"],
        "updated_at": run["updated_at"].isoformat(),
    }

    if run["status"] in {"queued", "running", "waiting_approval"}:
        hot = await read_hot_state(redis, run_id)
        if (
            hot is not None
            and hot.attempt_id == str(run["current_attempt_id"])
            and hot.fencing_token == run["fencing_token"]
        ):
            response["progress"] = hot.progress
            response["current_step"] = hot.current_step
            response["heartbeat_at"] = hot.updated_at

    return response

여기서 Redis timeout은 짧아야 한다. 실시간 투영값을 잠시 사용할 수 없다고 전체 status endpoint가 30초씩 멈출 이유는 없다. Redis 읽기가 실패하면 진행률을 생략하고 영속적 state를 반환하는 제한 모드를 둘 수 있다.

반대로 종료 status에서는 PostgreSQL 결과가 기준이다. Redis에 아직 running이 남아 있어도 UI는 영속적 succeeded를 우선해야 한다.

실패 경계가 저장 위치의 답을 검증한다

실패	PostgreSQL	Redis	Checkpoint	기대 동작
Worker 프로세스 충돌	running일 수 있음	heartbeat가 오래됨	마지막 snapshot 유지	Queue 재전달 후 재개
Redis 전체 손실	유지	사라짐	영속적 저장소라면 유지	진행률 일부 손실, run/result 보존
이전 worker의 늦은 쓰기	최신 attempt 유지	stale update 도착	최신 checkpoint 유지	fencing token으로 거부
PostgreSQL 사용 불가	읽기·종료 commit 불가	실시간 state만 남을 수 있음	저장 불가	성공으로 완료 처리하지 않음
배포로 schema 변경	유지	투영값 재생성 가능	이전 schema snapshot	Versioned loader로 변환

실패복구 행동남아 있어야 할 근거

• Worker crashcheckpoint에서 재개run record · checkpoint
• Redis lossprojection 재생성run record · checkpoint
• Stale workerattempt / fencing으로 쓰기 거부최신 run record
• PostgreSQL outage성공 처리·queue ack 보류Redis만으로 완료 금지
• Schema changeversioned loader로 이전 snapshot 변환versioned checkpoint

이 표에서 중요한 행은 PostgreSQL을 사용할 수 없는 경우다. 도구 실행은 끝났지만 최종 결과를 commit하지 못했다면 worker는 메시지를 성공으로 확인 처리해서는 안 된다. 그렇지 않으면 계산은 끝났지만 공식 기록에는 결과가 없는 상태가 된다.

부수 효과가 있는 도구가 포함되면 재시도가 같은 작업을 다시 실행할 수 있으므로 멱등성이 필요하다. 승인·멱등성·재시도는 Agent Tool은 정확히 한 번 실행될 수 있을까에서 다룬다.

경계를 무너뜨리는 안티패턴

1. 모든 state를 Python dictionary에 둔다

로컬 데모에는 충분하지만 프로세스 재시작과 다중 worker 실행을 견디지 못한다. 메모리에는 다시 만들 수 있는 객체만 둔다.

2. PostgreSQL JSONB 하나에 모든 것을 넣는다

Schema 변경은 쉬워 보이지만 자주 갱신되는 row의 경합, 문서 증가, 쿼리 복잡도가 커진다. 자주 조회하는 식별자와 status는 column으로, 반복되는 개체는 table로 분리한다.

3. Redis가 빠르다는 이유로 final result의 유일한 복사본을 둔다

TTL, eviction, 영속성 정책이 업무 보존 정책과 충돌할 수 있다. Redis를 기준 저장소로 쓸 것이라면 캐시와 다른 운영 보장을 명시해야 한다.

4. Redis HSET을 last-write-wins로 사용한다

재시도와 병렬 분기에서 오래된 진행률이 최신 state를 덮는다. 단일 쓰기 주체, version, 실행 시도 fencing 가운데 필요한 경계를 둔다.

5. TTL 없는 temporary key를 만든다

완료된 run의 실시간 state가 영구히 쌓인다. 쓰기와 TTL을 같은 transaction/script 안에 두고 종료 후 보존 기간을 별도로 정한다.

6. Redis state를 event history처럼 사용한다

Hash를 계속 덮어쓰면 현재 값은 알 수 있지만 어떻게 그 상태에 도달했는지는 사라진다. 반대로 모든 진행 event를 replay해서 현재 화면을 만들면 단순 조회가 비싸진다. State와 event는 다른 데이터 모델이다.

7. Checkpoint에 process object를 넣는다

Client, file handle, coroutine은 새 worker에서 복원할 수 없다. 직렬화 가능한 state와 자원 식별자만 저장하고 runtime 자원은 재생성한다.

Framework state와 application state를 구분한다

이 글의 구조는 특정 framework에 종속되지 않는다. LangGraph를 사용한다면 checkpoint 개념은 checkpointer에 대응한다. 공식 문서는 checkpointer가 thread의 graph-state snapshot을 저장하고, store는 thread 밖에서 애플리케이션이 정의한 장기 데이터를 저장한다고 구분한다. 메모리 내 saver는 프로세스 재시작 뒤 유지되지 않으므로 운영 환경에는 영속적 checkpointer가 필요하다고도 설명한다. (LangGraph Persistence)

이 글의 개념                    LangGraph에서의 mapping
───────────────────────────    ─────────────────────────────
Run의 durable business record   Application PostgreSQL tables
Resume 가능한 graph snapshot    Checkpointer
Cross-thread long-term memory    Store
현재 progress projection         Application-managed Redis state
Process-local temporary state    Runtime memory

thread_id를 넘기고 checkpointer를 연결했다고 run 기록, 권한 확인, UI 진행률, event 이력이 모두 해결되는 것은 아니다. Framework graph state와 애플리케이션 업무 state는 겹치는 부분이 있지만 같은 것은 아니다. 이 차이는 9편에서 더 자세히 다룬다.

위치보다 먼저 복구 계약을 정한다

State를 어디에 둘지는 속도 비교로 끝나지 않는다. 어떤 독자가 무엇을 묻는지, 어떤 실패 뒤에도 답이 남아야 하는지, 사라진 값을 다시 만들 수 있는지를 먼저 정해야 한다. 그 답이 저장 위치뿐 아니라 version, TTL, transaction과 보존 기간을 결정한다.

PostgreSQL
  durable truth: run, identity, status, result, audit

Redis Hash
  hot projection: current step, progress, heartbeat

Checkpoint
  resume snapshot: values, next nodes, completed work

Process memory
  disposable runtime objects

이 경계가 분명하면 Redis를 잃어도 run과 결과를 설명할 수 있고, worker를 잃어도 checkpoint에서 재개할 수 있으며, 오래된 worker가 최신 결과를 되돌리지 못한다.

다만 “지금 상태가 무엇인가”를 저장하는 것과 “무슨 일이 일어났는가”를 기록하는 것은 다른 문제다.

• 한 단계가 끝났다는 사실을 한 번만 전달하려면?
• 사용자가 재연결했을 때 놓친 진행률을 어떻게 다시 받는가?
• 현재 snapshot과 과거에 일어난 일의 목록을 같은 Redis Hash로 처리해도 되는가?

다음 글 State는 Event가 아니다에서는 현재 snapshot과 추가 전용 event 이력을 분리하고, Redis Hash와 Redis Stream이 각각 어떤 접근 패턴을 푸는지 살펴본다.