Agent가 일하면서 말하게 하기: SSE, Generator, Redis Streams

Agent에게 기업 분석을 요청하고 40초 동안 로딩 표시만 본다고 해보자. 최종 답이 좋아도 사용자는 시스템이 멈췄는지, 자료를 찾는 중인지, 승인을 기다리는지 알 수 없다.

실시간 UI는 그 빈 시간을 다음처럼 바꾼다.

연결됨
✓ 사내 문서 검색 완료
✓ 재무자료 정규화 완료
→ Valuation 계산 중
! 비용 1,000달러 초과 action 승인 필요

이전 글 Agent Tool은 정확히 한 번 실행될 수 있을까에서는 승인 대기, 실행, outcome_unknown 같은 Tool action 상태를 안전하게 다루는 방법을 정리했다. State는 Event가 아니다에서 만든 Redis snapshot·Stream cursor 위에, 이제 그 event와 action 진행 상태를 HTTP 연결로 브라우저에 전달하는 문제가 남는다.

여기서 세 개념이 자주 섞인다.

Async generator   Python이 값을 하나씩 생산하는 방법
StreamingResponse Python 값을 HTTP body chunk로 보내는 server adapter
SSE               Chunk 안에서 event를 표현하는 text protocol

Async generator가 있다고 자동으로 네트워크 stream이 되는 것은 아니다. SSE도 generator의 다른 이름이 아니다. 각 층이 다른 책임을 가진다.

먼저 무엇을 streaming하는지 구분한다

“Agent streaming”은 최소 세 가지를 가리킬 수 있다.

Stream	예시	특성
진행 event	research_completed, approval_required	낮은 빈도, replay 가치가 큼
출력 token	에, 이, 전, 트	높은 빈도, 부분 출력
최종 결과	완성된 보고서	한 번의 영속적 결과

이 글의 중심은 진행 event streaming이다. Redis Stream에 남아 재연결 후 replay할 수 있고, event type이 UI의 시간순 목록을 만든다. Token streaming은 훨씬 높은 빈도와 다른 buffering 정책이 필요하며 21편에서 별도로 다룬다.

Redis에 남기는 event 양과 보존

Redis Stream은 전달 계층이다. Worker가 발행하고 API가 cursor부터 읽어 SSE로 보낸다. 운영 원칙은 간단하다.

• Stream entry에는 UI·재연결에 필요한 작은 payload만 넣는다. prompt 전문, 대형 tool output, 감사에 필요한 전체 기록은 넣지 않는다.
• XTRIM, MAXLEN, MINID 등으로 보존 기간과 길이에 상한을 둔다. “모든 event를 Redis에 영구 보관”은 메모리 비용과 eviction 위험을 키운다.
• 장기 event 이력·감사·재현은 PostgreSQL table, 별도 log pipeline, 객체 저장소로 분리한다. Redis는 reconnect와 짧은 replay를 돕고, system of record가 아니다.
• Hash snapshot과 마찬가지로, Stream이 비어 있거나 trim되었어도 run 존재·권한·최종 결과는 PostgreSQL을 기준으로 한다.

Iterator는 protocol이고 generator는 구현 방법이다

Python에서 iterator는 “다음 값을 요청할 수 있는 객체”라는 protocol이다. __iter__()와 __next__()를 직접 구현할 수 있다.

class StepIterator:
    def __init__(self, steps: list[str]):
        self.steps = steps
        self.index = 0

    def __iter__(self):
        return self

    def __next__(self) -> str:
        if self.index >= len(self.steps):
            raise StopIteration

        value = self.steps[self.index]
        self.index += 1
        return value

for step in StepIterator(["research", "valuation", "report"]):
    print(step)

Generator는 yield를 사용해 iterator를 쉽게 만드는 Python 메커니즘이다.

def steps():
    yield "research"
    yield "valuation"
    yield "report"

yield가 실행되면 함수는 지역 변수와 실행 위치를 보존한 채 멈춘다. 다음 값을 요청받으면 그 지점부터 이어간다.

모든 generator는 iterator다.
모든 iterator가 generator인 것은 아니다.

Python 공식 용어로 iterator는 __next__()가 값을 반환하다 소진되면 StopIteration을 발생시키는 객체다. Async iterator는 __aiter__()와 awaitable을 반환하는 __anext__()를 구현하고, 종료 시 StopAsyncIteration을 사용한다. (Python Glossary)

네트워크 event에는 async generator가 자연스럽다

동기 generator는 다음 값을 만들기 위해 차단 I/O를 기다리면 현재 thread를 막는다. Redis에서 event가 생길 때까지 비차단 방식으로 기다리려면 async generator가 잘 맞는다.

async def run_events(redis, stream_key: str, cursor: str):
    while True:
        response = await redis.xread(
            streams={stream_key: cursor},
            count=100,
            block=15_000,
        )

        if not response:
            continue

        _, entries = response[0]
        for event_id, fields in entries:
            cursor = event_id
            yield event_id, fields

이 함수에는 await와 yield가 함께 있다.

• await redis.xread(...): 다음 event를 기다리는 동안 event loop에 실행권을 돌려준다.
• yield event: 준비된 event 하나를 소비자에게 넘기고 generator 실행을 잠시 멈춘다.
• 소비자가 다음 값을 요청해야 loop가 다시 진행된다.

Python 공식 문서는 async def 안에 yield가 있는 함수를 asynchronous generator function으로 정의한다. 호출하면 async generator iterator가 만들어지고, async for가 __anext__()의 awaitable을 반복해서 처리한다. (Asynchronous Generator Functions)

async for event_id, fields in run_events(redis, stream_key, cursor):
    print(event_id, fields)

여기까지는 Python 객체가 Python 객체를 소비하는 코드다. 아직 브라우저로 아무 byte도 가지 않았다.

yield를 HTTP body에 연결하는 역할이 필요하다

FastAPI의 StreamingResponse는 동기 iterator나 async iterator를 받아 각 항목을 응답 body 조각으로 전송한다. (StreamingResponse)

from fastapi.responses import StreamingResponse

async def chunks():
    yield b"first chunk\n"
    yield b"second chunk\n"

response = StreamingResponse(
    chunks(),
    media_type="text/plain",
)

관계를 그리면 다음과 같다.

Redis XREAD
    │ await
    ▼
async generator
    │ yield str/bytes
    ▼
StreamingResponse / ASGI server
    │ HTTP body chunks
    ▼
Browser

StreamingResponse가 모든 조각을 메모리에 모아 완성된 응답을 만드는 것이 아니다. Iterator가 값을 만들 때마다 전송할 기회를 제공한다. 다만 실제 packet 경계와 브라우저가 읽는 시점은 ASGI 서버, HTTP stack, proxy와 buffering의 영향을 받는다. yield 한 번이 네트워크 packet 하나를 보장하지는 않는다.

SSE는 HTTP 응답 안의 framing 규칙이다

SSE 응답의 media type은 text/event-stream이다. 각 event는 UTF-8 텍스트 field로 구성되고 빈 줄로 끝난다.

id: 1760000000123-0
event: dcf_started
data: {"scenario":"base"}

주요 field는 다음과 같다.

Field	역할
data	Client에 전달할 payload
event	사용자 정의 event type
id	재연결 cursor로 사용할 마지막 event ID
retry	User agent의 재연결 지연 힌트
:로 시작하는 줄	주석. Heartbeat로 활용 가능

여러 줄의 data: field는 줄바꿈으로 결합된다. 이 글에서는 JSON을 한 줄로 인코딩해 event 하나를 만든다.

import json

def encode_sse(
    *,
    event_id: str | None = None,
    event_type: str | None = None,
    data: dict | None = None,
    retry_ms: int | None = None,
) -> str:
    lines: list[str] = []

    if event_id is not None:
        if "\n" in event_id or "\r" in event_id:
            raise ValueError("invalid SSE event id")
        lines.append(f"id: {event_id}")

    if event_type is not None:
        if "\n" in event_type or "\r" in event_type:
            raise ValueError("invalid SSE event type")
        lines.append(f"event: {event_type}")

    if retry_ms is not None:
        lines.append(f"retry: {retry_ms}")

    if data is not None:
        payload = json.dumps(data, separators=(",", ":"))
        lines.append(f"data: {payload}")

    return "\n".join(lines) + "\n\n"

def encode_heartbeat() -> str:
    return ": heartbeat\n\n"

Event type과 ID에 줄바꿈을 허용하지 않는 것은 framing 삽입 공격을 막기 위해서다. Payload는 직접 문자열 보간하지 않고 JSON encoder를 사용한다.

WHATWG HTML Standard는 SSE event stream 형식, id, 재연결 시간과 Last-Event-ID 동작을 정의한다. (Server-Sent Events)

Snapshot과 실시간 event를 한 generator에서 연결한다

7편에서 Redis Hash의 last_event_id가 snapshot과 일관되게 저장된다고 가정했다. 첫 연결에서는 snapshot을 먼저 보내고 그 cursor 이후 event를 읽는다.

import asyncio
from collections.abc import AsyncIterator

TERMINAL_EVENTS = {
    "run_succeeded",
    "run_failed",
    "run_cancelled",
}

async def sse_event_stream(
    *,
    request,
    redis,
    run_id: str,
    resume_cursor: str | None,
) -> AsyncIterator[str]:
    state_key = f"agent:run:{{{run_id}}}:state"
    stream_key = f"agent:run:{{{run_id}}}:events"

    if resume_cursor is None:
        snapshot = await redis.hgetall(state_key)
        if not snapshot:
            yield encode_sse(
                event_type="resync_required",
                data={"reason": "snapshot_missing"},
            )
            return

        cursor = snapshot.get("last_event_id", "0-0")
        snapshot_status = snapshot["status"]
        yield encode_sse(
            event_id=cursor,
            event_type="snapshot",
            data={
                "status": snapshot_status,
                "current_step": snapshot["current_step"],
                "progress": int(snapshot["progress"]),
            },
            retry_ms=2_000,
        )

        if snapshot_status in {"succeeded", "failed", "cancelled"}:
            return
    else:
        cursor = resume_cursor

    try:
        while True:
            if await request.is_disconnected():
                return

            response = await redis.xread(
                streams={stream_key: cursor},
                count=100,
                block=15_000,
            )

            if not response:
                yield encode_heartbeat()
                continue

            _, entries = response[0]
            for event_id, fields in entries:
                cursor = event_id
                event_type = fields["type"]

                yield encode_sse(
                    event_id=event_id,
                    event_type=event_type,
                    data={
                        "type": event_type,
                        "occurred_at": fields["occurred_at"],
                        "payload": json.loads(
                            fields.get("payload_json", "{}")
                        ),
                    },
                )

                if event_type in TERMINAL_EVENTS:
                    return
    except asyncio.CancelledError:
        # Client disconnect나 server shutdown의 cancellation을 삼키지 않는다.
        raise

이 generator는 세 가지 상황에서 끝난다.

• 요청 연결이 끊어졌다.
• 종료 event를 전달했다.
• ASGI 서버가 작업을 취소했다.

CancelledError를 잡고 무시하면 종료와 구조화된 취소가 깨질 수 있다. Python 공식 문서는 정리가 필요하면 try/finally를 사용하고, CancelledError를 명시적으로 잡았다면 일반적으로 정리 후 다시 전파하라고 권장한다. (Task Cancellation)

Redis 연결이나 다른 자원을 generator 내부에서 직접 열었다면 finally에서 반환해야 한다.

async def stream_with_dedicated_client(redis_pool, **kwargs):
    redis = await redis_pool.acquire()
    try:
        async for chunk in sse_event_stream(redis=redis, **kwargs):
            yield chunk
    finally:
        await redis_pool.release(redis)

실제 redis-py pool API는 사용하는 version과 wrapper에 맞춰 적용한다. 핵심은 오래 지속되는 응답이 연결 하나를 오래 점유한다는 사실을 용량 계획에 포함하는 것이다.

Endpoint는 streaming을 시작하기 전에 실패할 일을 끝낸다

HTTP 응답 header가 전송된 뒤에는 일반 JSON 404나 401로 바꿀 수 없다. 따라서 인증, 권한 확인, run 존재 확인과 입력 검증은 StreamingResponse를 반환하기 전에 끝낸다.

from fastapi import Depends, Header, HTTPException, Request
from fastapi.responses import StreamingResponse

@app.get("/runs/{run_id}/events")
async def stream_run_events(
    run_id: str,
    request: Request,
    last_event_id: str | None = Header(
        default=None,
        alias="Last-Event-ID",
    ),
    user: User = Depends(current_user),
):
    # PostgreSQL lookup이 끝나면 DB transaction/connection은 반환한다.
    run = await run_repository.get_for_tenant(
        run_id=run_id,
        tenant_id=user.tenant_id,
    )
    if run is None:
        raise HTTPException(404, "run not found")

    generator = sse_event_stream(
        request=request,
        redis=redis,
        run_id=run_id,
        resume_cursor=last_event_id,
    )

    return StreamingResponse(
        generator,
        media_type="text/event-stream",
        headers={
            "Cache-Control": "no-cache, no-transform",
            "X-Accel-Buffering": "no",
        },
    )

X-Accel-Buffering은 일부 Nginx 배포에서 buffering을 끄기 위한 구현별 header다. 모든 proxy가 이해하는 표준 header는 아니다. 실제 ingress, CDN, load balancer가 streaming 응답을 buffer하는지 배포 환경에서 확인해야 한다.

PostgreSQL 연결을 SSE 수명 내내 잡고 있지 않은 점도 중요하다. 권한 확인 쿼리를 마친 뒤 연결을 pool에 반환하고, 오래 지속되는 loop에는 Redis 읽기 연결만 사용한다.

첫 연결과 재연결을 구분한다

첫 연결에서는 snapshot과 last_event_id를 보낸다. 브라우저가 SSE id를 처리한 뒤 네트워크가 끊기면 재연결 요청에 Last-Event-ID를 포함할 수 있다. Server는 그 cursor 이후 Redis Stream entry를 읽는다.

첫 연결
  snapshot(id=42) → event 43 → event 44

disconnect

재연결
  Last-Event-ID: 44
  XREAD after 44 → event 45 ...

재연결 요청에 cursor가 있는데 다시 최신 snapshot을 보내면 이전 event replay와 snapshot 순서가 혼란스러울 수 있다. 기본 동작을 명확히 나눈다.

• Cursor 없음: snapshot을 보내고 그 cursor 이후 실시간 stream 시작
• Cursor 있음: cursor validity를 확인하고 그 이후부터 replay
• Cursor가 trimming 범위보다 오래됨: resync_required를 보내거나 snapshot endpoint 재호출 유도

7편에서 구현한 cursor_is_too_old() 검사를 endpoint 시작 전에 적용할 수 있다. Header가 임의 문자열일 수 있으므로 Stream ID 문법 검증도 필요하다.

import re

STREAM_ID_PATTERN = re.compile(r"^[0-9]+-[0-9]+$")

def validate_stream_cursor(value: str | None) -> str | None:
    if value is None:
        return None
    if not STREAM_ID_PATTERN.fullmatch(value):
        raise HTTPException(400, "invalid Last-Event-ID")
    return value

Heartbeat는 애플리케이션 event가 아니다

Agent가 30초 동안 LLM 응답을 기다리면 Stream에 새 event가 없을 수 있다. 그동안 proxy나 브라우저는 연결이 유휴 상태라고 판단할 수 있다.

XREAD BLOCK 15000이 timeout되면 SSE comment를 보낸다.

: heartbeat

콜론으로 시작하는 주석 줄은 브라우저 애플리케이션 event 처리기에 전달되지 않지만 byte는 연결을 통과한다. Heartbeat는 다음 역할을 한다.

• 유휴 연결에 트래픽을 만든다.
• Server가 주기적으로 연결 해제를 확인할 기회를 준다.
• 네트워크 경로가 끊겼을 때 쓰기 실패를 발견할 수 있다.

Heartbeat 간격은 proxy 유휴 timeout보다 짧아야 하지만 너무 잦으면 연결 수만큼 불필요한 트래픽이 생긴다. 15초는 예시일 뿐 배포 환경에 맞춰 측정해야 한다.

Heartbeat를 Redis Stream에 XADD하지 않는 것도 중요하다. 연결 상태 신호는 도메인 event가 아니다. Viewer가 1,000명이면 각 연결이 독립적으로 heartbeat를 보내면 되고, 공유 event 이력을 오염시킬 이유가 없다.

연결 해제는 취소 protocol의 시작점이다

브라우저가 탭을 닫았다고 Agent run까지 취소해야 하는 것은 아니다.

SSE connection lifecycle       Agent run lifecycle
─────────────────────────      ─────────────────────────
tab close → connection end     worker는 계속 실행 가능
network reconnect              같은 run cursor부터 재구독
explicit cancel button         별도 durable cancellation request

SSE generator 취소는 전송 작업을 멈추는 것이다. Worker의 Agent 실행 취소는 별도 API와 영속적 state 전환이 필요하다. 둘을 연결하면 사용자가 Wi-Fi를 바꾸는 것만으로 run이 취소되는 문제가 생긴다.

13편에서 취소를 별도 protocol로 다루는 이유다.

Backpressure는 slow client를 인정하는 데서 시작한다

생산자가 1초에 event 100개를 만들고 client가 1초에 10개밖에 처리하지 못한다면 어디엔가 적체가 생긴다.

나쁜 구현은 Redis event를 백그라운드 작업으로 무제한 미리 읽어 프로세스 메모리 queue에 쌓는다.

# 피할 것: client 속도와 무관하게 memory queue가 계속 커질 수 있다.
async def prefetch_forever(redis, queue, stream_key, cursor):
    while True:
        events = await redis.xread({stream_key: cursor}, count=1000)
        for event in events:
            await queue.put(event)

앞의 async generator는 소비자가 다음 조각을 요청해야 다음 loop로 진행한다. StreamingResponse가 이전 조각 전송을 await하는 동안 generator는 무제한으로 Redis를 읽지 않는다. 이것이 자연스러운 pull 기반 backpressure의 시작점이다.

다만 애플리케이션 수준의 yield가 모든 네트워크 버퍼를 직접 제어하는 것은 아니다. ASGI 서버와 kernel socket buffer가 제한된 양을 보관할 수 있다. 다음 경계도 필요하다.

• XREAD COUNT로 replay 묶음 크기 제한
• 한 연결이 따라잡을 수 있는 최대 지연 정의
• Cursor가 보존 범위 밖이면 snapshot 재동기화
• 느린 연결의 timeout 또는 최대 수명
• Event payload 크기 제한
• 연결 수에 맞는 Redis pool 용량

SSE client 하나가 Redis 차단 연결 하나를 점유하는 구조는 작은 규모에서는 단순하지만 viewer 수가 커지면 비싸다. 그때는 프로세스 안에서 run별 reader 하나가 event를 읽고 여러 연결에 fan-out하는 계층을 고려할 수 있다. 다만 fan-out 계층 자체의 queue 제한과 느린 구독자 제거가 새로 필요하다.

Stream이 시작된 뒤의 오류는 HTTP status로 바꿀 수 없다

Generator가 세 번째 event를 보내다 Redis 오류를 만나면 응답은 이미 200 OK로 시작되었다. 이제 503 JSON 응답으로 교체할 수 없다.

선택지는 세 가지다.

1. 일시적 오류를 제한된 횟수와 기한 안에서 재시도한다.
2. 가능하면 SSE stream_error 제어 event를 보낸 뒤 닫는다.
3. 즉시 연결을 닫고 client 재연결에 맡긴다.

어느 경우든 서버 로그에는 run ID, 마지막 cursor, 오류 class를 남겨야 한다. Client가 재연결하면 같은 cursor부터 replay할 수 있어야 한다.

async def safe_sse_stream(**kwargs):
    try:
        async for chunk in sse_event_stream(**kwargs):
            yield chunk
    except asyncio.CancelledError:
        raise
    except RedisConnectionError:
        logger.exception(
            "run event stream failed",
            extra={"run_id": kwargs["run_id"]},
        )
        yield encode_sse(
            event_type="stream_error",
            data={"retryable": True},
            retry_ms=2_000,
        )

마지막 yield도 네트워크가 이미 끊겼다면 전달되지 않을 수 있다. 제어 event는 보조 신호일 뿐, 복구의 근거는 Redis Stream cursor다.

브라우저에서는 EventSource로 소비한다

브라우저의 기본 EventSource API는 사용자 정의 event type을 addEventListener로 받는다.

const source = new EventSource(`/runs/${runId}/events`);

source.addEventListener("snapshot", (message) => {
  const snapshot = JSON.parse(message.data);
  renderSnapshot(snapshot);
});

source.addEventListener("dcf_started", (message) => {
  const event = JSON.parse(message.data);
  timeline.upsert(message.lastEventId, event);
});

source.addEventListener("run_succeeded", (message) => {
  const event = JSON.parse(message.data);
  timeline.upsert(message.lastEventId, event);
  source.close();
  refreshFinalResult();
});

source.onerror = () => {
  showConnectionState("reconnecting");
};

기본 EventSource는 일반 fetch처럼 임의 요청 header를 자유롭게 설정하는 API가 아니다. 동일 출처 보안 cookie를 사용할지, 수명이 짧은 서명 URL을 사용할지, fetch 기반 SSE client를 사용할지는 인증 아키텍처에 맞춰 정해야 한다. 수명이 긴 access token을 query string에 그대로 넣으면 브라우저 이력과 proxy 로그에 남을 수 있으므로 피한다.

교차 출처 연결이라면 CORS와 자격 증명 정책도 명시적으로 구성해야 한다. 권한 확인은 연결 시작 시 한 번만 하는 것으로 충분한지도 검토한다. 권한이 run 도중 취소될 수 있다면 연결 수명 또는 주기적 재검증 정책이 필요하다.

SSE, polling, WebSocket 중 무엇을 고를까

요구사항	Polling	SSE	WebSocket
Server → 브라우저 갱신	가능	자연스러움	가능
브라우저 → server 실시간 메시지	요청마다	별도 HTTP API	양방향
자동 재연결	직접 구현	브라우저 지원	직접 protocol 설계
HTTP infrastructure 친화성	높음	높음, buffering 확인	Upgrade 지원 필요
Event ID/replay	직접 구현	Protocol cursor와 결합 쉬움	직접 구현
단순 진행률 UI	가능하지만 지연·요청 증가	좋은 기본 선택	대개 과함

Agent 진행률은 대부분 서버에서 client로 향하는 단방향이다. 사용자의 승인, 취소, 추가 입력은 일반 POST endpoint로 보낼 수 있다. 이 경우 SSE가 단순하다.

WebSocket이 나쁜 것이 아니다. 지연 시간이 짧은 양방향 제어, 음성, 협업 상호작용처럼 양방향 메시지가 지속적으로 필요하면 더 적합할 수 있다. 전송 방식 선택은 “Agent니까 WebSocket”이 아니라 통신 형태로 결정한다.

운영 환경에서 자주 놓치는 경계

Proxy buffering

애플리케이션은 조각을 yield했는데 proxy가 여러 조각을 모아 한 번에 전달할 수 있다. Localhost에서 잘 보인다고 운영 환경에서도 같은 주기를 보장하지 않는다. Ingress와 CDN 설정을 실제 streaming endpoint로 검증한다.

Compression buffering

Middleware나 proxy 압축이 작은 조각을 buffer할 수 있다. text/event-stream을 압축 대상에서 제외할지 환경별로 확인한다.

Idle timeout

Load balancer와 proxy timeout보다 짧은 heartbeat가 필요하다. 모든 구간의 제한을 확인한다.

Connection limits

열린 SSE 연결 수만큼 file descriptor, 애플리케이션 작업, socket buffer와 경우에 따라 Redis 연결을 사용한다. 초당 요청 수만 보고 용량을 정하면 안 된다.

Graceful shutdown

배포 시 새 연결 수락을 중단하고, 기존 stream에 재연결 가능한 cursor가 있다는 전제 아래 연결을 닫는다. SSE 연결을 영원히 비우려고 하면 배포가 끝나지 않는다.

Observability

활성 연결 수, 연결 지속 시간, 재연결 비율, heartbeat 쓰기 실패, cursor 지연, Redis XREAD 지연 시간을 측정한다. HTTP 200 수만 세면 stream이 실제 event를 전달했는지 알 수 없다.

End-to-end mental model

Worker
  │ HSET + XADD
  ▼
Redis Hash + Stream
  │ HGETALL / XREAD(cursor)
  ▼
Async generator
  │ yield SSE frames
  ▼
StreamingResponse / ASGI
  │ chunked HTTP body
  ▼
EventSource
  │ event ID + type + JSON data
  ▼
Browser state + timeline

각 층의 실패도 다르다.

계층	실패 예시	복구 기준
Worker	프로세스 충돌	Queue와 checkpoint
Redis projection	Key expiry	PostgreSQL snapshot fallback
Redis Stream	Cursor trim	Snapshot resync
API 서버	배포/재시작	Client 재연결
네트워크	연결 해제	마지막 event ID replay
브라우저	탭 새로고침	Snapshot + cursor 초기화

이 구조에서 SSE 서버는 Agent 실행을 소유하지 않는다. Redis event를 cursor부터 읽어 전달하는 gateway다. 브라우저 연결이 사라져도 worker는 계속 실행하고, 다시 연결된 브라우저는 snapshot과 보존된 event로 UI를 복원한다.

구현 확인 목록

• Iterator, async generator, HTTP streaming과 SSE를 구분했는가?
• 인증과 권한 확인을 응답 시작 전에 끝내는가?
• PostgreSQL 연결을 stream 수명 동안 점유하지 않는가?
• 첫 연결은 snapshot cursor에서 시작하는가?
• 재연결은 Last-Event-ID 이후를 읽는가?
• Cursor 문법과 보존 범위를 검증하는가?
• Redis XREAD에 block timeout과 count limit가 있는가?
• 빈 읽기 결과에는 heartbeat를 보내는가?
• CancelledError를 삼키지 않는가?
• Client 연결 해제가 Agent run 취소로 이어지지 않는가?
• 종료 event 뒤 stream을 닫는가?
• 느린 client 때문에 메모리 queue가 무한히 커지지 않는가?
• Proxy buffering, 압축과 유휴 timeout을 운영 환경에서 검증했는가?
• EventSource authentication에 장기 token query parameter를 사용하지 않는가?

이제 Agent는 실행 중인 일을 브라우저에 말할 수 있다. 하지만 worker 안에서 실제 graph runtime이 어떤 state 갱신과 custom event를 내보내고, 그것이 애플리케이션 state와 어떻게 다른지는 아직 남아 있다.

다음 글 LangGraph는 Architecture가 아니다: Runtime의 정확한 위치에서는 graph.astream(), updates, custom, reducer와 checkpoint를 살펴보고, LangGraph state·Redis state·browser state를 구분한다.