Streaming, hủy run, drain

Tổng quan

Làm cho run quan sát được theo thời gian thực mà không gắn việc thực thi vào HTTP request. Lưu mọi event của run vào một bảng run_events chỉ-append, fan-out các event trực tiếp qua một broadcaster trong bộ nhớ, và phơi GET /v1/runs/{id}/events dưới dạng SSE có resume bằng Last-Event-ID (replay từ bảng, rồi gắn vào luồng trực tiếp). Thêm việc hủy run, các lane của scheduler có drain, goroutine được recover panic, và graceful shutdown biết về SSE. Client có thể ngắt kết nối rồi gắn lại mà không mất một event nào.

Yêu cầu

Chức năng

• run_events: các dòng chỉ-append (run_id, seq) với kind, payload jsonb, created_at.
• Loop phát ra: run.started, turn.started, block.start, text.delta, thinking.delta, tool_input.delta, block.stop, tool.started, tool.finished, turn.finished, turn.aborted, usage, run.finished, error, stream_degraded.
• Chính sách lỗi của `EventSink.Emit`, nêu một lần và tuân thủ ở mọi nơi: log ở mức warn kèm run id và loại event, tăng `hango.run_events.write_failures`, retry đúng một lần, và khi thất bại lần hai thì phát một event đánh dấu `stream_degraded` (cố hết sức) rồi tiếp tục. **Một run không bao giờ bị làm thất bại chỉ vì không ghi được một event.** Vì vậy nơi gọi bỏ qua lỗi của `Emit`, trừ khi để lan truyền việc hủy context.
• Một DB poller (ticker `safego`, `HANGO_QUEUED_POLL_INTERVAL`, mặc định 5s) claim các run `queued` cũ hơn `HANGO_QUEUED_POLL_AGE` (mặc định 10s) và không nằm trong hàng đợi bộ nhớ, để một run bị mất enqueue (tiến trình chết giữa lúc insert và enqueue, hoặc hàng đợi đầy) vẫn được thực thi. Nó claim qua conditional claim của phase 5, nên không thể thực thi đôi.
• Gauge `hango.run.oldest_queued_age` (số giây kể từ lúc run `queued` cũ nhất được tạo) do poller phát ra; phase 14 đặt alert trên nó. Đó là metric duy nhất chứng minh các run không bị bỏ quên trong im lặng.
• GET /v1/runs/{id}/events stream text/event-stream với id: là seq của event, event: là loại event, data: là payload JSON.
• Header Last-Event-ID (hoặc ?last_event_id=) replay các event đã lưu có seq lớn hơn, rồi gắn vào luồng trực tiếp. Không khoảng trống, không trùng lặp. Replay là một vòng lặp, không phải một query: ListAfter(runID, cursor, limit) được gọi lặp lại cho đến khi cả hai điều kiện cùng đúng: trang trả về ngắn (ít dòng hơn limit) và seq cao nhất đã replay lớn hơn hoặc bằng seq đầu tiên được staged từ subscription trực tiếp. Cho đến lúc đó handler vẫn tiếp tục phân trang, vì một run bận rộn có thể lưu nhiều event hơn một trang chứa nổi trong lúc replay.
• Các event đến từ subscription trực tiếp *trong lúc* replay được staged vào một slice riêng có giới hạn byte (`HANGO_SSE_REPLAY_STAGE_BYTES`, mặc định 1MiB), không phải buffer giao hàng, để pha replay không bao giờ làm tràn buffer của subscriber và đánh rơi chính những event nó sắp cần. Nếu chạm giới hạn stage, handler bỏ việc gắn vào luồng trực tiếp, đóng bằng một event lỗi `reconnect` và để client resume từ `id:` cuối cùng của nó: tính đúng đắn đến từ replay, luôn luôn như vậy.
• `seq` được cấp phát **bên trong** cùng transaction chèn các dòng `run_events` (`UPDATE runs SET last_event_seq = last_event_seq + $n ... RETURNING` rồi insert nhiều dòng), nên một lần crash giữa lúc cấp phát và insert không thể để lại một lỗ hổng vĩnh viễn mà vòng lặp replay sẽ chờ mãi.
• Loại event `turn.aborted`: một lỗi provider giữa chừng stream phát `turn.aborted` mang theo message dở dang và đặt lại trạng thái chỉ số block của emitter trước mọi lần retry, để một lượt được retry bắt đầu lại chỉ số block từ 0 và client không bao giờ thấy hai event `block.start` cho chỉ số 0 mà không có một abort xen giữa.
• Keepalive là comment : ping mỗi HANGO_SSE_KEEPALIVE (mặc định 15s).
• Các trạng thái kết thúc của run đóng stream sau khi phát run.finished; run đã kết thúc thì replay toàn bộ log và đóng ngay.
• POST /v1/runs/{id}/cancel đặt cancel_requested_at, hủy context của run trong bộ nhớ nếu nó là local, và trả về 202; run kết thúc ở trạng thái cancelled với state dở dang đã được lưu.
• Lane của scheduler: các lane có tên (main bây giờ, subagent ở phase 9) với giới hạn concurrency độc lập, MarkDraining và Stats.
• Shutdown: đánh dấu draining, ngừng nhận run, hủy các SSE subscriber bằng một event server.draining, chờ các run đang hoạt động tối đa HANGO_DRAIN_TIMEOUT, rồi hủy chúng, lưu state dở dang và nhả lease để một tiến trình khác (hoặc chính tiến trình này khi restart) requeue chúng tại ranh giới lượt thay vì làm chúng thất bại. HANGO_DRAIN_TIMEOUT phải tối thiểu bằng một lượt đầy đủ (mặc định ghi trong tài liệu là 120s); phần validate config cảnh báo khi nó ngắn hơn.
• Vòng đời topic của broadcaster: một topic tồn tại cho đến khi run đạt trạng thái kết thúc **hoặc** subscriber cuối cùng rời đi, tùy cái nào đến sau; nó được đếm tham chiếu, không bị đóng ngay khi run kết thúc. Do đó một run bị tạm dừng (phase 15) không giữ goroutine nào trong khi các SSE client vẫn có thể còn subscribe; giới hạn của việc này được ghi tài liệu và quan sát được qua `hango.sse.subscribers` và `hango.broadcaster.topics`. Mục của run trong cancel registry bị xóa ở đúng hai điểm: khi run đạt trạng thái kết thúc, và khi nó tạm dừng (phase 15). Cả hai nằm trong `defer` của worker.

Phi chức năng

• Các delta khối lượng lớn không được ghi một dòng cho mỗi token: emitter gộp (coalesce) delta thành các batch bị giới hạn theo thời gian/kích thước (HANGO_EVENT_FLUSH_INTERVAL mặc định 100ms, tối đa 8KiB) trước khi lưu, và mỗi batch đã flush chỉ được publish cho subscriber trực tiếp sau khi đã được lưu (bước 4 và test persist-before-publish của nó).
• http.NewResponseController(w).SetWriteDeadline được đặt lại cho mỗi lần ghi để không timeout nào của server giết một stream dài.
• Mọi goroutine trong hango đều được khởi động qua safego.Go, hàm này recover panic, ghi log stack và tăng một counter.
• Phát hiện rò rỉ dựa trên hành vi, không dựa trên profile: các test assert rằng `runtime.NumGoroutine()` trở về giá trị trước test (với một khoảng chờ ổn định có giới hạn) sau khi run hoàn tất, subscriber ngắt kết nối và hủy run, **và** `Broadcaster.TopicCount()` trở về không. Profile `goroutineleak` của Go 1.27 chỉ được dùng trong soak test của phase 16, nơi một tiến trình chạy lâu làm nó có ý nghĩa; nó không thấy được các rò rỉ mà những phase này thực sự có nguy cơ gặp (một goroutine bị chặn mãi mãi nhưng vẫn còn reachable).

Kiến trúc

// internal/safego/safego.go
func Go(ctx context.Context, log *slog.Logger, component, op string, fn func(context.Context))
func Recover(log *slog.Logger, component, op string) // defer safego.Recover(...)()

// internal/run/broadcaster.go
type Broadcaster struct{} // per-run topics, bounded per-subscriber buffers

type Subscription struct {
	Events <-chan store.RunEvent
	Cancel func()
}

func (b *Broadcaster) Publish(runID uuid.UUID, ev store.RunEvent)
func (b *Broadcaster) Subscribe(runID uuid.UUID, buffer int) *Subscription // slow consumer is dropped, not blocking
func (b *Broadcaster) CloseRun(runID uuid.UUID) // marks terminal; topic is freed when refcount hits 0
func (b *Broadcaster) TopicCount() int          // test + gauge. Red team: F11

// internal/run/emitter.go
type Emitter struct{} // implements agent.EventSink

func (e *Emitter) Emit(ctx context.Context, ev agent.Event) error // seq alloc + coalesce + persist + publish
func (e *Emitter) Flush(ctx context.Context) error                // called at turn boundaries and on finish

// internal/run/scheduler.go (extended)
type Lane struct {
	Name        string
	Concurrency int
}

func (s *Scheduler) MarkDraining()
func (s *Scheduler) Stats() map[string]LaneStats
func (s *Scheduler) Cancel(runID uuid.UUID) bool // true when the run was local and cancelled

// internal/server/sse.go
type SSEWriter struct{}

func (w *SSEWriter) Event(id int64, kind string, payload []byte) error // resets write deadline, flushes
func (w *SSEWriter) Ping() error

-- migrations/00004_run_events.sql
alter table runs add column last_event_seq bigint not null default 0; -- Red team: F8 — seq source

create table run_events (
  id uuid primary key,
  tenant_id uuid not null references tenants(id) on delete cascade,
  run_id uuid not null references runs(id) on delete cascade,
  seq bigint not null,
  kind text not null,
  payload jsonb not null,
  created_at timestamptz not null default now(),
  unique (run_id, seq)
);
create index run_events_tenant_run_seq_idx on run_events (tenant_id, run_id, seq);

Instrumentation

• SSE handler span: span SERVER sse.run_events với hango.run.id, sse.resume_from_seq, sse.replayed_events, sse.live_events, sse.close_reason (client_disconnect|run_finished|draining|lagged).
• Response header: response SSE mang X-Hango-Run-Id, X-Trace-Id (trace của SSE handler) và traceparent mang span context root của run, để client đối chiếu được stream của mình với run trace mà nó đang theo dõi.
• Span event trên root span của run: run.cancel_requested (kèm user đã yêu cầu), run.drained.
• Metric (tên trong internal/telemetry/metrics_names.go): gauge hango.sse.subscribers, gauge hango.broadcaster.topics, counter hango.sse.events_published (attribute kind), counter hango.sse.lagged_subscribers, histogram hango.run_events.rows theo từng run (giám sát write amplification), counter hango.run_events.write_failures, gauge hango.run.oldest_queued_age, histogram hango.sse.replay_pages, gauge hango.scheduler.draining.
• Trường log: emitter log ở mức debug cho mỗi lần flush với hango.run.id, kích thước batch và số byte; các lần drop và drain log ở mức warn.

Các file liên quan

Tạo mới

• internal/safego/safego.go, internal/safego/safego_test.go.
• internal/run/broadcaster.go, internal/run/broadcaster_test.go.
• internal/run/emitter.go, internal/run/emitter_test.go.
• internal/server/sse.go, internal/server/handlers_run_events.go, internal/server/sse_test.go.
• internal/store/pg/run_event.go, internal/store/pg/run_event_test.go.
• internal/run/leak_test.go — assert runtime.NumGoroutine() trước/sau cộng Broadcaster.TopicCount() == 0, cho run hoàn tất, client ngắt kết nối và hủy run.
• internal/run/poller.go, internal/run/poller_test.go — DB poller cho run queued và gauge tuổi của nó.
• internal/server/sse_headers_test.go — assert X-Hango-Run-Id, X-Trace-Id và một traceparent đúng định dạng trên response SSE.
• migrations/00004_run_events.sql.
• docs/learning/06-streaming-and-cancellation.md.

Sửa

• internal/agent/loop.go — phát event ở mọi ranh giới; phản hồi ctx.Err() kịp thời; kết thúc trên context.WithoutCancel.
• internal/run/scheduler.go, internal/run/worker.go — lane, draining, cancel registry, nối emitter.
• internal/server/handlers_runs.go — endpoint cancel; GET /v1/runs/{id} có thêm last_event_seq.
• internal/server/server.go — route SSE, không có write timeout, event draining khi shutdown.
• internal/app/app.go — broadcaster, emitter factory, thứ tự drain: HTTP draining → scheduler draining → chờ → cancel → đóng pool.
• internal/config/config.go — HANGO_SSE_KEEPALIVE, HANGO_SSE_WRITE_TIMEOUT, HANGO_SSE_REPLAY_STAGE_BYTES, HANGO_EVENT_FLUSH_INTERVAL, HANGO_EVENT_MAX_BATCH_BYTES, HANGO_DRAIN_TIMEOUT (≥ một lượt), HANGO_LANE_MAIN_CONCURRENCY, HANGO_QUEUED_POLL_INTERVAL, HANGO_QUEUED_POLL_AGE.
• internal/store/pg/tenancy_test.go — thêm case run_events.
• docs/system-architecture.md, docs/learning/README.md, README.md (ví dụ curl SSE).

Các bước triển khai

1. Viết internal/safego và chuyển một cách máy móc mọi go func() hiện có trong codebase sang safego.Go.
2. Viết migrations/00004_run_events.sql (thêm runs.last_event_seq và bảng run_events) và RunEventStore: Append(ctx, runID, events...) cấp phát seq bên trong cùng transaction với insert (UPDATE runs SET last_event_seq = last_event_seq + $n RETURNING, rồi insert nhiều dòng, rồi commit), và ListAfter(ctx, runID, afterSeq, limit).
3. Viết internal/run/broadcaster.go: map[uuid.UUID]*topic, mỗi topic được đếm tham chiếu theo subscriber và chỉ được giải phóng khi run đã kết thúc và refcount bằng không. Mỗi topic giữ một tập channel subscriber với buffer có giới hạn; khi buffer đầy, bỏ subscriber đó và đánh dấu nó lagged để SSE handler đóng kết nối bằng một event error bảo client kết nối lại với Last-Event-ID (đúng đắn nhờ replay, không bao giờ chặn loop). Phơi TopicCount().
4. Viết internal/run/emitter.go: cài đặt agent.EventSink (được định nghĩa trong internal/agent; phụ thuộc chạy một chiều). Các delta cùng loại và cùng chỉ số được cộng dồn trong một buffer, flush theo khoảng thời gian, theo kích thước, hoặc khi có bất kỳ event không phải delta; các event không phải delta flush ngay. Mỗi event đã flush được lưu rồi mới publish, theo đúng thứ tự đó, để subscriber không bao giờ thấy một event chưa thể resume. Áp dụng chính sách lỗi của Emit: log, đếm, retry một lần, rồi phát stream_degraded và tiếp tục, không bao giờ trả về một lỗi có thể làm run thất bại. Khi gặp turn.aborted, đặt lại trạng thái chỉ số block.
5. Mở rộng loop để phát toàn bộ từ vựng event. Các delta của provider ánh xạ một-một từ provider.Event; thêm tool.started/tool.finished mang tên tool, output ForUser và IsError.
6. Viết internal/server/sse.go: đặt Content-Type: text/event-stream, Cache-Control: no-cache, Connection: keep-alive, X-Accel-Buffering: no; mỗi lần ghi gọi rc.SetWriteDeadline(time.Now().Add(HANGO_SSE_WRITE_TIMEOUT)) rồi rc.Flush().
7. Viết handlers_run_events.go: kiểm tra quyền truy cập run theo scope của request (giới hạn theo creator đối với member, 404 khi bị từ chối); parse Last-Event-ID; subscribe trước và stage các event trực tiếp vào replay stage có giới hạn byte; rồi lặp ListAfter cho đến khi trang ngắn và seq cao nhất đã replay ≥ seq trực tiếp đầu tiên được staged; sau đó flush stage (bỏ qua các seq đã replay) và đọc subscription trực tiếp; select trên subscription, ticker keepalive, r.Context().Done() và channel draining của server. Ghi lại hango.sse.replay_pages. 7a. Viết internal/run/poller.go: một ticker safego chọn các run queued cũ hơn HANGO_QUEUED_POLL_AGE và không có trong hàng đợi bộ nhớ, enqueue chúng (khử trùng lặp làm việc này an toàn), và publish hango.run.oldest_queued_age.
8. Thêm POST /v1/runs/{id}/cancel: lưu cancel_requested_at, gọi scheduler.Cancel(runID), trả về 202 kèm run. Một run đang queued bị worker hủy tại thời điểm claim. Handler gọi SetAuditAction(c, "run.cancel", "run", runID), nên một run bị hủy luôn nêu tên người đã hủy.
9. Mở rộng loop để kiểm tra ctx.Err() giữa các lượt và sau mỗi tool call, và lưu assistant message dở dang trước khi trả về StatusCancelled.
10. Mở rộng scheduler với lane và MarkDraining; Enqueue trả về lỗi khi đã draining. Nối thứ tự shutdown trong internal/app.
11. Viết test: fan-out của broadcaster, bỏ subscriber chậm, và vòng đời topic có đếm tham chiếu (topic sống sót qua trạng thái kết thúc khi còn subscriber, và TopicCount() trở về không khi nó rời đi); gộp và thứ tự của emitter (assert persist-before-publish) cùng chính sách lỗi của nó (hai lần ghi thất bại liên tiếp sinh ra stream_degraded và một run succeeded); resume dưới tải: một run sinh ra nhiều event hơn một trang ListAfter trong lúc client kết nối lại, assert luồng nối lại bằng đúng log, mỗi event một lần (đây là test bắt được replay không phân trang); replay stage tràn thì đóng bằng reconnect thay vì đánh rơi; turn.aborted đặt lại chỉ số block qua một lần retry provider; test cancel; test drain assert một run bị drain được requeue tại ranh giới lượt chứ không bị fail; test poller (một run được chèn mà không enqueue vẫn thực thi); các assert số goroutine và số topic trước/sau.
12. Cập nhật docs; thêm một ví dụ SSE curl -N vào README.

Dàn ý tài liệu học

docs/learning/06-streaming-and-cancellation.md phải có:

• Khái niệm: vì sao run là đơn vị của tính bền vững (durability) còn SSE là một subscriber chứ không phải bên thực thi.
• Định dạng wire của SSE: các dòng id:/event:/data:/comment, Last-Event-ID, ngữ nghĩa kết nối lại, proxy buffering, và vì sao X-Accel-Buffering: no quan trọng.
• Lựa chọn thiết kế: persist-rồi-publish, subscribe-rồi-replay có khử trùng lặp, gộp delta để giảm write amplification, bỏ subscriber chậm thay vì chặn loop, đặt lại write-deadline bằng ResponseController, keepalive ping, cancel qua một cờ đã lưu cộng một context trong bộ nhớ, kết thúc trên context.WithoutCancel.
• Đánh đổi: SSE so với WebSocket so với long polling; broadcaster trong bộ nhớ so với listen/notify so với một queue (và chính xác thứ gì hỏng khi có hai replica).
• So sánh với goclaw: WebSocket RPC do các kênh chat điều khiển, các callback onChunk nổi lên thành AgentEvent, FinalizeStage trên background context, activeRuns atomic cho drain, MarkDraining trong lane manager, safego.Recover.
• Cách debug tầng này: “client không thấy gì” được trả lời bằng span sse.run_events (nó có subscribe không, resume_from_seq là bao nhiêu, close_reason là gì) cộng hango.sse.lagged_subscribers; “event bị trùng” được trả lời bằng cách so khoảng seq của run_events đã lưu với số đếm replayed_events/live_events của handler; một lần shutdown bị treo được trả lời bằng hango.scheduler.draining và các span event run.drained.
• So sánh với deepagents: streaming là astream của LangGraph; việc tạm dừng và tiếp tục đi qua một checkpointer và thread_id, cùng một ý mà hango cài đặt bằng run bền vững cộng một event log.
• Vì sao hango đi khác: chỉ dùng SSE, một event log đã lưu làm nguồn sự thật cho replay (goclaw không thể resume một stream đã đứt).
• Vì sao "replay rồi gắn vào" là một vòng lặp bắt kịp (catch-up) chứ không phải một query, và vì sao các event trực tiếp phải được staged bên ngoài buffer giao hàng trong lúc replay.
• Vì sao rò rỉ goroutine ở đây phát hiện tốt nhất bằng cách đếm, không phải bằng profile `goroutineleak`.
• Bài tập: kill curl giữa stream rồi gắn lại; làm một subscriber chậm bằng breakpoint của debugger và quan sát chính sách lag; chèn một run queued thẳng vào Postgres và xem poller nhặt nó lên.
• Ngân sách thời gian: 14h cài đặt + 2h tài liệu học.

Tiêu chí thành công

• go build ./... && go vet ./... && go test ./... đều qua.
• Test resume SSE: ngắt kết nối sau N event, kết nối lại với Last-Event-ID: N, và luồng nối lại bằng toàn bộ event log đúng một lần, kể cả biến thể trong đó run phát ra nhiều event hơn một trang replay trong lúc client đang kết nối lại.
• Endpoint SSE của một run đã kết thúc replay toàn bộ log và đóng mà không bị treo.
• Test cancel: run đạt cancelled, run.finished được phát, assistant message dở dang được lưu, và một dòng audit_events với action run.cancel nêu tên user đã quyết định.
• Test drain: SIGTERM trong lúc có run đang hoạt động cho ra succeeded, hoặc một run được requeue có lease đã được nhả và resume_turn là lượt hoàn tất cuối cùng của nó, không bao giờ là một dòng bị mất và không bao giờ là một run failed chỉ vì shutdown; tiến trình thoát trong vòng HANGO_DRAIN_TIMEOUT + 5s.
• Số goroutine trở về giá trị trước test và Broadcaster.TopicCount() trở về không sau khi run hoàn tất, client ngắt kết nối và hủy run.
• Hai lần ghi run_events thất bại liên tiếp sinh ra stream_degraded và một run vẫn đạt succeeded; hango.run_events.write_failures tăng.
• Một run được chèn thẳng vào database mà không enqueue được poller nhặt lên trong vòng HANGO_QUEUED_POLL_INTERVAL + HANGO_QUEUED_POLL_AGE, và hango.run.oldest_queued_age trở về gần không.
• Gộp delta: một response fake 2000 token sinh ra ít hơn 50 dòng run_events (có assert).
• go test ./internal/store/pg/ -run TestCrossTenant bao gồm run_events.
• Response SSE mang X-Hango-Run-Id, X-Trace-Id và một traceparent trỏ vào root span của run (có test).
• Trong Grafana, thấy một span sse.run_events với các attribute resume và close-reason, và các series hango.sse.subscribers/hango.sse.events_published tồn tại.

Đánh giá rủi ro

Rollback

goose down một bước sẽ xóa run_events và cột runs.last_event_seq. Gỡ route SSE đưa client trở lại việc poll của phase 5, vốn vẫn hoạt động vì dòng run vẫn là state có thẩm quyền.