State는 Event가 아니다: 실시간 Agent UI 설계법

이전 글 Agent State는 어디에 있어야 하는가에서는 PostgreSQL에 영속적 run 기록과 재시작 기준 snapshot을, Redis Hash에는 UI·운영용 hot projection을, checkpoint에 재개용 graph snapshot을 두었다. Hash field는 PostgreSQL row를 대체하지 않는다.

사용자가 Agent run 화면을 연다고 하자. 브라우저는 먼저 세 가지에 답해야 한다.

• 지금 무엇을 하고 있으며 얼마나 진행됐는가?
• 지금까지 사용자에게 의미 있는 어떤 일이 일어났는가?
• 연결이 끊겼다가 돌아오면 어디서부터 이어 받아야 하는가?

이 질문은 각각 snapshot, event timeline, delivery cursor가 답한다.

Snapshot
  현재 상태: 규칙 적용 중 65%
  현재 작업: 예외 규칙 4개 중 3개 검토

Timeline
  ✓ 문서 수집 완료
  ✓ 입력 정규화 완료
  → 규칙 적용 중
  ! 승인 필요

Cursor
  last_event_id = 1760000000123-0

app.example.com/runs/7d5

문서 검토 run출장비 예외 검토

●실시간 연결됨

현재 상태규칙 적용 중

65%방금 갱신

61% → 62% 같은 연속 변화는 이 카드만 갱신하며 timeline 항목을 만들지 않는다.

현재 작업 · Snapshot

예외 규칙 4개 중 3개 검토

적용 조직

APAC / Korea

기준 시점

2026-10-05

다음 행동

담당자 승인 대기

활동 · Event timeline

1. 09:41
   문서 수집 완료12개 문서 · 최신본 확인
2. 09:42
   입력 정규화 완료금액·날짜·조직 범위
3. 09:43
   규칙 적용 중4개 중 3개 검토
4. 대기
   승인 필요예외 적용은 담당자 확인 필요

재연결 상태도 화면 계약이다연결 복구 중 → cursor replay → 이력 만료 시 snapshot 재동기화

Hash field를 계속 덮어쓰면 현재 값만 남는다. 반대로 event만 쌓으면 화면을 열 때마다 처음부터 모두 replay해서 현재 진행률을 계산해야 한다. 둘은 같은 데이터를 중복 저장하는 실수가 아니라, current state, event history, delivery stream이라는 서로 다른 층의 의도적인 투영값이다.

Redis Stream은 delivery stream에 가깝다. 실시간 fan-out, reconnect cursor, 짧은 replay 범위에는 적합하다. 반면 장기 감사 원장이나 “법무·보안이 몇 년 뒤에도 반드시 읽어야 하는 event log”의 유일한 저장소로 쓰면 안 된다. Stream trim·TTL·메모리 압력은 운영 정책의 일부다. 감사·재현·분쟁 대응에 필요한 이력은 PostgreSQL table, append-only log store, 객체 저장소처럼 영속 계층에 별도로 남긴다. Stream entry가 사라져도 run 결과와 checkpoint는 PostgreSQL에 남아야 한다.

브라우저 화면은 snapshot과 timeline을 함께 읽는다

구분	State	Event
질문	지금 어떤 상태인가?	어떤 일이 일어났는가?
쓰기 방식	같은 field를 덮어씀	새 entry를 추가
읽기 방식	최신 snapshot 한 번	Cursor 이후 범위
대표 구조	Redis Hash	Redis Stream
예시	progress=65	rules_applied
보존	짧은 TTL 가능	Replay 범위만큼 보존
UI 역할	즉시 화면 복원	시간순 목록과 점진적 갱신

핵심 차이는 시간이다.

State
time ──────────────────────────────────────────────>
                   { progress: 65, step: apply_rules }
                   현재 한 점

Events
time ──────────────────────────────────────────────>
     documents_collected ─ inputs_normalized ─ rules_applied ─ ...
                   변화의 순서

진행률 65는 event가 아니다. 시간이 지나 70으로 바뀌면 이전 값은 보통 필요 없다. 반면 approval_required는 사용자의 행동과 감사에 의미가 있는 도메인 event다. Event type을 progress_changed_to_61, progress_changed_to_62처럼 남발하면 시간순 목록이 잡음으로 가득 찬다.

좋은 시작점은 다음과 같다.

• State에는 자주 바뀌는 수치형 진행률과 현재 단계를 둔다.
• Event에는 사용자가 이해할 수 있는 단계 전환과 중요한 도메인 사실을 남긴다.
• Token 하나마다 event를 만들지 않는다. Token streaming은 별도의 대용량 채널로 본다.

브라우저 영역읽는 계약표시하는 정보

• Status headersnapshotstatus · progress · current_step
• Current worksnapshotactive item · approval state
• Timelinerecent events단계 전환 · 승인 · 복구 · 완료
• Live updatescursor + SSElast_event_id 이후 event

Snapshot은 화면을 복원하고, event는 timeline을 이어가며, cursor는 둘 사이의 경계를 고정한다.

저장 표현은 Redis Hash와 Stream으로 나뉜다

Article 5의 key naming을 이어가면 한 run에 두 key가 생긴다.

agent:run:{run_id}:state    Redis Hash
agent:run:{run_id}:events   Redis Stream

중괄호 안의 run_id는 Redis Cluster hash tag다. 두 key를 같은 slot에 배치하면 transaction이나 Lua script에서 함께 다룰 수 있다.

State Hash는 작고 현재적이다.

HSET agent:run:{7d5}:state
  status          running
  current_step    apply_rules
  progress        65
  last_event_id   1760000000123-0
  updated_at      2026-10-04T05:10:00Z

Event Stream은 추가 전용 시간순 기록이다.

1760000000010-0  documents_collected
1760000000060-0  inputs_normalized
1760000000123-0  rules_started

Redis Stream entry는 ID와 field-value 쌍으로 구성된다. XADD에 *를 사용하면 Redis가 단조 증가 ID를 생성한다. 일반적인 ID 모양은 <millisecondsTime>-<sequenceNumber>지만 애플리케이션은 이를 업무 timestamp로 계산하기보다 opaque cursor로 취급하는 편이 안전하다. (XADD)

Event schema부터 고정한다

Redis Stream은 schema를 강제하지 않는다. 그렇기 때문에 애플리케이션이 안정적인 공통 형식을 정의해야 한다.

from dataclasses import dataclass, field
from datetime import UTC, datetime
from typing import Any

@dataclass(frozen=True)
class RunEvent:
    type: str
    run_id: str
    attempt_id: str
    fencing_token: int
    payload: dict[str, Any] = field(default_factory=dict)
    schema_version: int = 1
    occurred_at: str = field(
        default_factory=lambda: datetime.now(UTC).isoformat()
    )

Event type은 UI 문장이 아니라 안정적인 기계 판독용 이름으로 둔다.

좋음
documents_collected
rules_applied
approval_required
run_succeeded

피할 것
문서를 모두 불러왔습니다!
검토가 거의 끝났어요

표시 문구와 번역은 client가 event type과 payload를 바탕으로 만든다. 그래야 문구가 바뀌어도 과거 event schema가 깨지지 않는다.

occurred_at과 Stream ID도 구분해야 한다.

• Stream ID는 Redis 안에서 순서와 cursor에 쓴다.
• occurred_at은 생산자가 기록한 도메인 timestamp다.
• 여러 생산자의 시계가 완벽히 일치한다고 가정하지 않는다.
• UI 순서는 한 stream 안에서는 Stream ID를 우선한다.

XADD로 bounded event history를 만든다

간단한 redis-py 비동기 코드는 다음처럼 시작할 수 있다.

import json
from redis.asyncio import Redis

MAX_EVENTS_PER_RUN = 1_000

def run_events_key(run_id: str) -> str:
    return f"agent:run:{{{run_id}}}:events"

async def append_run_event(
    redis: Redis,
    event: RunEvent,
) -> str:
    event_id = await redis.xadd(
        run_events_key(event.run_id),
        fields={
            "type": event.type,
            "attempt_id": event.attempt_id,
            "fencing_token": event.fencing_token,
            "schema_version": event.schema_version,
            "occurred_at": event.occurred_at,
            "payload_json": json.dumps(event.payload),
        },
        maxlen=MAX_EVENTS_PER_RUN,
        approximate=True,
    )
    return event_id

이 단독 XADD helper는 snapshot을 바꾸지 않는 event 형식을 설명하기 위한 최소 예시다. 여러 attempt가 경쟁할 수 있는 worker에서 직접 호출하려면 현재 fencing token을 원자적으로 검증해야 한다. Snapshot과 timeline을 함께 바꾸는 실제 전환에는 아래의 fenced publish_transition을 사용한다.

MAXLEN ~ 1000에 해당하는 근사 trimming을 함께 요청했다. 정확한 trimming은 임계값을 정확히 맞추지만 더 많은 작업이 필요할 수 있다. 근사 trimming은 내부 구조 단위로 효율적으로 제거하므로 실제 길이가 임계값보다 잠시 클 수 있다. (XTRIM)

Event 수 기준이 아니라 시간 기준 보존이 필요하면 MINID 전략을 고려할 수 있다. 어떤 전략이든 “모든 event를 영구 보존”하는 것이 기본값이어서는 안 된다. Redis Stream은 UI replay 범위를 위한 운영 로그이고, 장기 감사가 필요하면 PostgreSQL이나 객체 저장소에 별도로 보존한다.

하나의 전환은 snapshot과 event를 원자적으로 바꾼다

다음 두 명령을 별도로 호출하면 중간 실패가 생긴다.

await redis.hset(state_key, mapping={"progress": 65})
await redis.xadd(events_key, {"type": "rules_started"})

첫 번째 명령 뒤에 worker가 죽으면 state는 65%인데 대응하는 event는 없다. 반대로 event부터 쓰고 state 갱신이 실패하면 timeline에는 규칙 적용이 시작됐지만 snapshot은 이전 단계에 머문다.

모든 state 갱신에 event가 필요한 것은 아니다. 하지만 하나의 도메인 전환이 snapshot과 시간순 목록을 함께 바꾼다면 둘을 같은 원자적 작업으로 묶는 것이 좋다.

Auto-generated Stream ID를 state의 last_event_id에도 저장해야 하므로 Lua script가 편리하다. Script 안에서 XADD 결과를 받아 바로 HSET할 수 있다.

PUBLISH_TRANSITION = """
local current_fence = tonumber(redis.call('HGET', KEYS[1], 'fencing_token'))
local proposed_fence = tonumber(ARGV[4])

if current_fence and proposed_fence < current_fence then
    return redis.error_reply('stale fencing token')
end

local event_id = redis.call(
    'XADD', KEYS[2], 'MAXLEN', '~', ARGV[1], '*',
    'type', ARGV[2],
    'attempt_id', ARGV[3],
    'fencing_token', ARGV[4],
    'schema_version', ARGV[5],
    'occurred_at', ARGV[6],
    'payload_json', ARGV[7]
)

redis.call(
    'HSET', KEYS[1],
    'attempt_id', ARGV[3],
    'fencing_token', ARGV[4],
    'status', ARGV[8],
    'current_step', ARGV[9],
    'progress', ARGV[10],
    'last_event_id', event_id,
    'updated_at', ARGV[6]
)

redis.call('EXPIRE', KEYS[1], ARGV[11])
redis.call('EXPIRE', KEYS[2], ARGV[11])
return event_id
"""

async def publish_transition(
    redis: Redis,
    *,
    event: RunEvent,
    status: str,
    current_step: str,
    progress: int,
    ttl_seconds: int,
) -> str:
    result = await redis.eval(
        PUBLISH_TRANSITION,
        2,
        f"agent:run:{{{event.run_id}}}:state",
        run_events_key(event.run_id),
        MAX_EVENTS_PER_RUN,
        event.type,
        event.attempt_id,
        event.fencing_token,
        event.schema_version,
        event.occurred_at,
        json.dumps(event.payload),
        status,
        current_step,
        progress,
        ttl_seconds,
    )
    return result

이 script는 stale worker의 token을 먼저 거부한 뒤 event와 snapshot을 함께 바꾼다. 같은 attempt 안의 세부 순서까지 경쟁할 수 있다면 선행 글처럼 version 비교도 추가한다. 이 원자성은 Redis 안의 로컬 일관성일 뿐 PostgreSQL run 기록까지 포함하는 분산 transaction은 아니다. 종료 결과는 PostgreSQL에 먼저 commit하고, 그다음 Redis projection과 종료 event를 갱신한다. Redis 갱신이 실패하면 영속적 state에서 다시 조정할 수 있어야 한다.

last_event_id는 snapshot boundary다

실시간 연결을 단순하게 구현하면 다음 race가 발생한다.

1. Client가 HGETALL로 snapshot을 읽음
2. Worker가 event E를 publish함
3. Client가 XREAD '$'로 새 event만 기다림

결과: E는 snapshot 뒤, subscribe 전에 발생해 놓칠 수 있음

$는 XREAD 호출 시점의 stream 끝에서 시작해 이후 새 entry만 기다리는 의미다. Snapshot과 $ 사이에 생긴 event는 전달되지 않는다.

State와 event를 atomic하게 갱신하면서 state에 last_event_id를 넣으면 연결 순서를 다음처럼 만들 수 있다.

1. HGETALL state
   → snapshot + last_event_id = 1760000000123-0

2. 필요하면 XRANGE로 최근 history를 last_event_id까지 조회

3. XREAD from 1760000000123-0
   → 그 ID 이후 event를 replay하고 계속 대기

Snapshot 이후 event가 발생하면 그 ID는 snapshot cursor보다 크므로 XREAD가 읽는다. Snapshot 전에 발생한 event는 이미 snapshot에 반영되어 있고, timeline이 필요하면 XRANGE로 조회한다.

Bootstrap은 화면 복원에 필요한 계약을 한 번에 반환한다

UI가 처음 열리면 현재 state와 최근 event 몇 개를 반환한다.

from dataclasses import dataclass

@dataclass(frozen=True)
class RunBootstrap:
    durable_status: str
    state: dict[str, str] | None
    events: list[tuple[str, dict[str, str]]]
    cursor: str | None
    replay_complete: bool

async def load_run_bootstrap(
    redis: Redis,
    *,
    run_id: str,
    durable_status: str,
    history_limit: int = 50,
) -> RunBootstrap:
    state_key = f"agent:run:{{{run_id}}}:state"
    events_key = run_events_key(run_id)

    state = await redis.hgetall(state_key)
    if not state:
        return RunBootstrap(
            durable_status=durable_status,
            state=None,
            events=[],
            cursor=None,
            replay_complete=False,
        )

    cursor = state.get("last_event_id", "0-0")

    # 최신 history를 역순으로 읽고 UI에는 시간순으로 돌려준다.
    recent_desc = await redis.xrevrange(
        events_key,
        max=cursor,
        min="-",
        count=history_limit,
    )
    recent = list(reversed(recent_desc))

    return RunBootstrap(
        durable_status=durable_status,
        state=state,
        events=recent,
        cursor=cursor,
        replay_complete=True,
    )

RunBootstrap은 Redis 응답이 아니라 브라우저와 API 사이의 계약이다. 권한 확인과 durable_status 조회는 이 함수보다 먼저 PostgreSQL에서 수행한다. replay_complete=false면 client는 timeline이 완전하다고 표시하지 않고 durable 결과 또는 degraded active 화면을 사용한다. Event payload에는 prompt, 도구 인자, PII, 자격 증명 대신 UI에 필요한 최소 field만 발행한다.

XREAD로 cursor 이후 event를 읽는다

Bootstrap 이후에는 마지막 cursor부터 새 entry를 읽는다.

async def read_next_events(
    redis: Redis,
    *,
    run_id: str,
    last_event_id: str,
    block_ms: int = 15_000,
    count: int = 100,
) -> list[tuple[str, dict[str, str]]]:
    response = await redis.xread(
        streams={run_events_key(run_id): last_event_id},
        count=count,
        block=block_ms,
    )

    if not response:
        return []

    _, entries = response[0]
    return entries

XREAD에 전달한 ID는 배타적 cursor다. 176...-0을 전달하면 그 ID보다 큰 entry를 반환한다. BLOCK은 새 데이터가 없을 때 일정 시간 기다리게 하며, timeout 후 빈 응답을 반환할 수 있다. 이 timeout은 SSE heartbeat를 보내거나 client 연결 해제를 확인할 기회를 만든다. (XREAD)

COUNT도 중요하다. 재연결 뒤 event가 많이 밀렸다면 한 번에 무제한으로 가져오지 않고 묶음으로 처리한다. 다음 묶음의 cursor는 실제로 전달을 마친 마지막 event ID로 갱신한다.

Redis Stream ID를 SSE id로 전달한다

SSE event는 텍스트 형식이며 각 메시지를 빈 줄로 구분한다. Redis Stream ID를 SSE의 id field로 그대로 사용하면 저장소 cursor와 브라우저 재연결 cursor가 하나로 연결된다.

id: 1760000000123-0
event: rules_started
data: {"run_id":"7d5...","rules_total":4}

간단한 encoder는 다음과 같다.

import json

def encode_sse(event_id: str, fields: dict[str, str]) -> str:
    payload = {
        "type": fields["type"],
        "occurred_at": fields["occurred_at"],
        "payload": json.loads(fields.get("payload_json", "{}")),
        "schema_version": int(fields.get("schema_version", 1)),
    }

    return (
        f"id: {event_id}\n"
        f"event: {fields['type']}\n"
        f"data: {json.dumps(payload)}\n\n"
    )

HTML Standard의 SSE 모델은 event에 id를 지정할 수 있고, user agent가 재연결할 때 마지막 event ID를 사용하도록 정의한다. (Server-Sent Events)

하지만 이 글에서는 SSE endpoint 전체를 만들지 않는다. Async generator, StreamingResponse, heartbeat comment, proxy buffering, 연결 해제 취소는 다음 8편에서 다룬다. 여기서 중요한 것은 SSE가 Redis Stream의 cursor를 인식하는 event를 전달하는 전송 방식이라는 점이다.

재연결은 replay이며 replay는 중복을 허용해야 한다

네트워크는 정확히 메시지 경계에서만 끊기지 않는다.

Server: event 42를 socket에 씀
Network: browser가 받음
Browser: 처리 직후 connection 끊김
Server: browser가 처리했는지 알 수 없음
Browser: last ID 41로 reconnect할 수도 있음

이 경우 event 42가 다시 전달될 수 있다. 따라서 UI handler는 event ID를 기준으로 idempotent하게 동작해야 한다.

const seenEventIds = new Set();

function applyRunEvent(message) {
  if (seenEventIds.has(message.lastEventId)) return;
  seenEventIds.add(message.lastEventId);

  const event = JSON.parse(message.data);
  timeline.upsert(message.lastEventId, event);
}

실제 브라우저 메모리에 모든 ID를 영원히 보관할 필요는 없다. 시간순 항목의 key로 event ID를 사용하거나 마지막으로 적용한 cursor와 순서를 비교할 수 있다. 중요한 원칙은 “SSE는 정확히 한 번 전달된다”고 가정하지 않는 것이다.

State 갱신도 멱등이어야 한다. Event payload에 절대 진행률 65를 보내면 중복 적용해도 결과가 같다. progress += 5 같은 증분만 보내면 replay 시 70이 되는 문제가 생긴다.

Cursor가 보존 범위보다 오래되면 snapshot으로 돌아간다

Stream을 trim하면 오래된 client의 cursor가 이미 삭제되었을 수 있다. Redis는 cursor ID 자체가 없어도 그보다 큰 보존 entry를 읽을 수 있지만, 삭제된 구간의 event는 복구할 수 없다.

이때 조용히 일부 이력만 보내면 UI가 불완전한 시간순 목록을 완전한 것으로 오해한다. Server는 cursor가 보존된 가장 오래된 ID보다 오래됐는지 확인하고 재동기화를 요구해야 한다.

def parse_stream_id(value: str) -> tuple[int, int]:
    milliseconds, sequence = value.split("-", 1)
    return int(milliseconds), int(sequence)

async def cursor_is_too_old(
    redis: Redis,
    *,
    run_id: str,
    cursor: str,
) -> bool:
    first = await redis.xrange(
        run_events_key(run_id),
        min="-",
        max="+",
        count=1,
    )
    if not first:
        return False

    oldest_id, _ = first[0]
    return parse_stream_id(cursor) < parse_stream_id(oldest_id)

Cursor가 너무 오래됐다면 protocol은 다음 중 하나를 선택할 수 있다.

• SSE에 resync_required 제어 event를 보내고 연결을 닫는다.
• Client가 GET /runs/{id} 초기화 endpoint를 다시 호출한다.
• Server가 새 snapshot event를 보내고 cursor를 최신 경계로 이동한다.

가장 단순한 시작은 snapshot 재조회 후 최신 cursor에서 재연결이다. 시간순 목록의 오래된 일부가 반드시 필요하다면 Redis 보존 기간을 늘리거나 영속적 event 보관소를 별도로 둔다.

Snapshot 자체가 사라진 경우도 정해야 한다

6편처럼 Redis Hash에는 TTL이 있을 수 있다. Run이 오래전에 끝나 state key와 stream key가 모두 만료되었다면 PostgreSQL의 영속적 기록을 대신 사용한다.

Redis state 있음
  → snapshot + events + live SSE

Redis state 없음, PostgreSQL run 있음
  → durable terminal result 반환
  → live stream은 열지 않음

Redis state 없음, PostgreSQL active run
  → worker heartbeat/queue ownership 확인
  → projection 재생성 또는 degraded response

Redis에서 찾지 못했다고 곧바로 404 run not found로 해석하면 안 된다. Run의 존재와 권한은 PostgreSQL이 결정한다.

Fan-out 대안도 UI의 소비 의미에 맞춰 고른다

Pub/Sub만 사용하면 왜 reconnect가 어려운가

Redis Pub/Sub는 실시간 fan-out에는 간단하다. 하지만 구독자가 연결되어 있지 않을 때 발행된 메시지를 나중에 다시 읽을 수 없다. Redis 공식 문서는 Pub/Sub를 최대 한 번 전달로 설명하며, 네트워크 연결 해제 등으로 구독자가 처리하지 못한 메시지는 다시 전달되지 않는다고 명시한다. (Redis Pub/Sub)

Pub/Sub
publish ───────────────> connected subscribers
           disconnect 구간의 message는 사라짐

Stream
XADD ──> retained entries
XREAD(last_id) ──> disconnect 구간을 replay

따라서 재연결과 replay가 필요한 Agent UI에는 Stream이 자연스럽다. Pub/Sub를 알림 용도로 추가하고 실제 payload는 Stream에서 읽는 혼합 방식도 가능하지만, 처음부터 둘을 모두 도입할 필요는 없다. XREAD BLOCK만으로도 대기 중인 reader를 깨우고 새 entry를 전달할 수 있다.

Consumer group은 SSE fan-out의 기본 선택이 아니다

Redis Stream consumer group은 메시지를 여러 consumer에게 분배하고 확인 처리와 대기 중인 작업의 소유권을 추적할 때 유용하다. Queue worker에는 잘 맞는다.

하지만 같은 run을 보고 있는 browser 두 개는 둘 다 같은 event를 받아야 한다.

SSE viewer A ─┐
              ├─ 둘 다 rules_started를 받아야 함
SSE viewer B ─┘

두 viewer를 같은 consumer group에 넣으면 event가 group consumer 사이에 분배되어 한쪽만 받을 수 있다. SSE viewer는 각자 cursor를 가진 XREAD reader로 시작하는 편이 맞다. Worker 작업 queue와 UI event stream이 모두 Redis Stream을 쓴다고 해서 소비 패턴까지 같지는 않다.

Stream 하나를 run마다 만들 것인가

이 글의 예시는 run마다 Stream key 하나를 둔다.

agent:run:{run_id}:events

장점은 권한 확인 이후 한 run만 읽기 쉽고, TTL과 trimming도 run 수명 주기에 맞출 수 있다는 점이다. 단점은 run 수만큼 key 수가 늘어난다는 것이다.

공용 Stream 하나에 모든 run event를 넣을 수도 있다.

agent:events
  run_id=7d5 type=rules_started
  run_id=91a type=documents_collected

하지만 XREAD는 임의 field 필터를 제공하지 않으므로 특정 run만 보려는 SSE 연결도 관계없는 event를 계속 읽고 애플리케이션에서 걸러야 한다. Tenant별 stream이나 분할 stream을 설계할 수 있지만 순서, 집중된 partition, 보존 정책이 복잡해진다.

초기 규모에서는 run별 Stream이 개념 모형과 보안 경계가 단순하다. 실제 key 수와 연결 수를 측정한 뒤 분할 방식을 바꾼다.

State와 event를 함께 발행하는 worker 코드

Worker node가 끝날 때 도메인 전환을 발행하는 흐름을 합치면 다음과 같다.

async def execute_rule_review_step(
    *,
    redis: Redis,
    run_id: str,
    attempt_id: str,
    fencing_token: int,
    state: dict,
) -> dict:
    await publish_transition(
        redis,
        event=RunEvent(
            type="rules_started",
            run_id=run_id,
            attempt_id=attempt_id,
            fencing_token=fencing_token,
            payload={"rules_total": 4},
        ),
        status="running",
        current_step="apply_rules",
        progress=65,
        ttl_seconds=24 * 60 * 60,
    )

    decision = await apply_policy_rules(state)
    next_state = {**state, "rule_decision": decision}

    await publish_transition(
        redis,
        event=RunEvent(
            type="rules_applied",
            run_id=run_id,
            attempt_id=attempt_id,
            fencing_token=fencing_token,
            payload={
                "rules_checked": decision.rules_checked,
                "requires_approval": decision.requires_approval,
            },
        ),
        status="running",
        current_step="approval",
        progress=80,
        ttl_seconds=24 * 60 * 60,
    )

    return next_state

Event payload에 규칙 적용 결과 전체를 복사하지 않았다. UI가 timeline을 그리는 데 필요한 작은 metadata만 넣고, 실제 결과는 checkpoint나 PostgreSQL record에 둔다. Stream entry가 작아야 replay와 메모리 사용량을 제어하기 쉽다.

Progress event와 domain event를 분리한다

모든 progress 변화가 timeline item이어야 하는 것은 아니다.

변화	State update	Event append
Progress 65 → 66	O	보통 X
Current step 변경	O	O
Tool retry 1회	O 가능	사용자에게 의미 있으면 O
Approval 필요	O	O
Token 한 개 생성	별도 channel	X
Run 성공	O	O

진행률은 연속적인 투영값이고 event는 의미 있는 전환이다. 이 기준이 없으면 LLM token, 내부 디버그 로그, 사용자용 event가 한 stream에 섞인다. 트래픽 특성과 보존 정책이 완전히 다른 데이터를 같은 채널로 보내면 느린 client와 메모리 압박을 다루기 어려워진다.

특히 token streaming은 진행률 streaming과 다르다. Token은 대용량 부분 출력이고 rules_applied 같은 event는 소량의 제어 신호다. 후속 streaming 글에서 이 차이를 별도로 다룬다.

브라우저에 보이는 것과 시스템 내부에서만 필요한 것도 나눈다.

채널	포함할 것	화면에서의 위치
Timeline event	단계 완료, 승인 요청, 사용자 개입, 복구·실패, 최종 완료	활동 timeline
State only	연속 progress, heartbeat, 현재 attempt의 내부 상태	Status card 또는 숨김
별도 token channel	생성 중인 부분 출력	결과 panel
Debug / audit	상세 trace, 원본 tool 인자, 장기 보존 기록	운영·감사 시스템

Event type은 화면 문구가 아니다. Client는 rules_applied와 작은 payload를 localization map에 통과시켜 “규칙 검토 완료”처럼 표시한다. Stream에 한국어 문장 자체를 저장하면 문구 수정과 다국어 UI가 event schema 변경으로 번진다.

Event sourcing 전체를 도입할 필요는 없다

State와 event를 분리한다고 모든 애플리케이션 state를 event replay로만 재구성해야 하는 것은 아니다.

이 글의 구조는 full event sourcing보다 작다.

PostgreSQL     durable business truth
Redis Hash    disposable current projection
Redis Stream  bounded operational event history
Checkpoint    resumable execution snapshot

Redis Stream을 지워도 PostgreSQL 결과와 checkpoint가 남고, Hash를 다시 만들 수 있다. Event는 UI와 관측성에 유용하지만 모든 업무 불변 조건의 유일한 기준 정보는 아니다.

완전한 event sourcing은 감사, 시간 쿼리, 결정론적 재구축 같은 강한 요구가 있을 때 검토할 수 있다. 단지 실시간 진행률 UI가 필요하다는 이유만으로 시스템 전체를 event 기반 아키텍처로 바꿀 필요는 없다.

브라우저는 bootstrap에서 시작해 replay 또는 resync로 돌아온다

최종적으로 브라우저가 run 화면을 여는 흐름은 다음과 같다.

Browser                  API/SSE server                 Redis
   │                           │                          │
   │ GET /runs/{id}/bootstrap  │                          │
   ├──────────────────────────>│ HGETALL state            │
   │                           ├─────────────────────────>│
   │                           │ XREVRANGE recent events   │
   │                           ├─────────────────────────>│
   │ snapshot + history+cursor │                          │
   │<──────────────────────────┤                          │
   │                           │                          │
   │ SSE connect(last cursor)  │                          │
   ├──────────────────────────>│ XREAD cursor BLOCK       │
   │                           ├─────────────────────────>│
   │ id + event + data         │<─────────────────────────┤
   │<──────────────────────────┤                          │
   │                           │                          │
   │ disconnect / reconnect    │ XREAD last-event-id      │
   ├──────────────────────────>├─────────────────────────>│

이 패턴의 중요한 성질은 세 가지다.

1. Snapshot 덕분에 event 전체를 replay하지 않고 화면을 즉시 그린다.
2. Cursor 덕분에 snapshot 이후 event를 놓치지 않는다.
3. Stream 보존 범위를 넘긴 재연결은 snapshot 재동기화로 복구한다.

1. 1Bootstrapsnapshot + recent events + cursor
2. 2Live SSEcursor 이후 event upsert
3. 3연결 끊김마지막 적용 ID 보존

Cursor가 replay 범위 안에 있는가?

예중복을 허용하며 이후 event replay

아니오resync_required → bootstrap 재호출

Redis miss는 404가 아니다. PostgreSQL durable record로 종료 결과 또는 degraded active response를 결정한다.

구현 전 확인 목록

• Current snapshot과 append-only event가 다른 key인가?
• State에 일관된 last_event_id가 저장되는가?
• Snapshot과 event 전환이 필요할 때 원자적으로 갱신되는가?
• Event envelope에 type, schema version, timestamp, attempt identity가 있는가?
• Snapshot과 event를 쓰기 전에 fencing token을 검증하는가?
• Redis Stream ID를 opaque cursor로 취급하는가?
• 재연결 시 마지막 event ID 이후를 replay하는가?
• 중복 event를 client가 안전하게 처리하는가?
• Cursor가 trimming 범위보다 오래됐을 때 재동기화 protocol이 있는가?
• Stream에 최대 길이 또는 시간 기반 보존 정책이 있는가?
• Redis에서 찾지 못할 때 PostgreSQL의 영속적 state를 대신 사용하는가?
• Event payload에서 자격 증명과 민감정보를 제거하는가?
• UI viewer에 consumer group을 잘못 사용하지 않는가?
• Progress, timeline, token, debug/audit channel을 화면과 보존 정책에 맞게 분리했는가?

이제 state와 event를 어디에 저장하고 어떤 cursor로 연결할지는 정했다. 그다음은 송금처럼 부수 효과가 있는 Tool이 Queue 재전달과 timeout 아래에서도 한 번의 사용자 의도로 수렴하게 만드는 일이다.

다음 글 Agent Tool은 정확히 한 번 실행될 수 있을까에서는 승인, action fingerprint, idempotency key와 outcome_unknown 처리를 다룬다. 그다음 Agent가 일하면서 말하게 하기에서 승인·실행·reconciliation 상태를 SSE로 전달한다.