Skip to content

feat: add async stream diagnostics #121

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
caigee-cmd merged 1 commit into from
Sep 9, 2026
Merged

feat: add async stream diagnostics #121

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
8 changes: 8 additions & 0 deletions CHANGELOG.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -7,8 +7,16 @@ Write each change in both `### English` and `### 中文` under `## Unreleased`.

### English

- Add asynchronous stream diagnostics for cancellation source, upstream status, SSE progress, and incomplete streams
- Classify client and request-context stream cancellations separately from upstream unavailability
- Show stream diagnostics in request details without storing prompt or response content

### 中文

- 异步记录流取消来源、上游状态、SSE 进度和未完成流等诊断信息
- 将客户端断开和请求上下文取消与上游不可用分开归类
- 请求详情展示流诊断，但不保存提示词或响应正文

## 0.3.6 - 2026-09-08

### English
Expand Down
17 changes: 17 additions & 0 deletions frontend/src/api/logs.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,6 +17,22 @@ export type RequestAttempt = {
usage_source?: string
}

export type RequestStreamDiagnostic = {
request_id: string
created_at: string
finished_at?: string | null
upstream_status?: number | null
upstream_request_id?: string
context_err?: string
cancellation_source?: string
relay_error?: string
sse_event_count: number
bytes_read: number
content_length: number
last_event?: string
saw_done: boolean
}

export type RequestLog = {
id: string
created_at: string
Expand All @@ -41,6 +57,7 @@ export type RequestLog = {
error_message?: string
attempt_count: number
attempts?: RequestAttempt[]
stream_diagnostic?: RequestStreamDiagnostic
}

export type RequestLogList = {
Expand Down
14 changes: 14 additions & 0 deletions frontend/src/i18n/messages.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -225,6 +225,13 @@ export const messages: Record<Lang, Dict> = {
logsRouting_sticky: 'Session sticky',
logsRouting_sticky_escape: 'Session escape',
logsNoAttempts: 'No attempt details.',
logsStreamDiagnostics: 'Stream diagnostics',
logsCancellationSource: 'Cancellation source',
logsUpstreamStatus: 'Upstream status',
logsSSEEvents: 'SSE events',
logsBytesRead: 'Bytes read',
logsLastEvent: 'Last event',
logsStreamComplete: 'Completed with [DONE]',
logsDetailTitle: 'Request detail',
logsShownTotal: '{shown} / {total}',
logsAutoRefresh: 'Live',
Expand Down Expand Up @@ -813,6 +820,13 @@ export const messages: Record<Lang, Dict> = {
logsRouting_sticky: '会话粘性',
logsRouting_sticky_escape: '粘性逃逸',
logsNoAttempts: '没有尝试详情。',
logsStreamDiagnostics: '流诊断',
logsCancellationSource: '取消来源',
logsUpstreamStatus: '上游状态',
logsSSEEvents: 'SSE 事件数',
logsBytesRead: '读取字节数',
logsLastEvent: '最后事件',
logsStreamComplete: '已收到 [DONE]',
logsDetailTitle: '请求详情',
logsShownTotal: '{shown} / {total}',
logsAutoRefresh: '实时',
Expand Down
26 changes: 26 additions & 0 deletions frontend/src/pages/LogsPage.tsx
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -816,6 +816,32 @@ export function LogsPage() {
{selected.error_message}
</div>
) : null}
{selected?.stream_diagnostic ? (
<div className="rounded-lg border border-separator bg-surface-secondary px-3 py-3 text-xs">
<div className="font-medium text-muted">{t('logsStreamDiagnostics')}</div>
<dl className="mt-2 grid gap-2 sm:grid-cols-2">
{[
[t('logsCancellationSource'), selected.stream_diagnostic.cancellation_source || '—'],
[t('logsUpstreamStatus'), selected.stream_diagnostic.upstream_status ?? '—'],
[t('logsSSEEvents'), selected.stream_diagnostic.sse_event_count],
[t('logsBytesRead'), selected.stream_diagnostic.bytes_read],
[t('logsLastEvent'), selected.stream_diagnostic.last_event || '—'],
[t('logsStreamComplete'), selected.stream_diagnostic.saw_done ? 'yes' : 'no'],
].map(([label, value]) => (
<div key={String(label)}>
<dt className="text-[10px] text-muted">{label}</dt>
<dd className="mono mt-0.5 break-all">{value}</dd>
</div>
))}
</dl>
{selected.stream_diagnostic.context_err || selected.stream_diagnostic.relay_error ? (
<div className="mt-2 space-y-1 break-all text-muted">
{selected.stream_diagnostic.context_err ? <div>context: {selected.stream_diagnostic.context_err}</div> : null}
{selected.stream_diagnostic.relay_error ? <div>relay: {selected.stream_diagnostic.relay_error}</div> : null}
</div>
) : null}
</div>
) : null}
<div>
<div className="text-xs font-medium text-muted">{t('logsAttempts')}</div>
{selected?.attempts?.length ? (
Expand Down
17 changes: 17 additions & 0 deletions internal/accounts/migrations.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -187,6 +187,23 @@ ALTER TABLE request_logs ADD COLUMN routing TEXT NOT NULL DEFAULT '';
CREATE INDEX IF NOT EXISTS request_logs_routing ON request_logs(routing);`},
{filename: "015_account_quota_state.sql", sql: `
ALTER TABLE accounts ADD COLUMN quota_json TEXT NOT NULL DEFAULT '';`},
{filename: "016_request_stream_diagnostics.sql", sql: `
CREATE TABLE IF NOT EXISTS request_stream_diagnostics (
request_id TEXT PRIMARY KEY REFERENCES request_logs(id) ON DELETE CASCADE,
created_at TEXT NOT NULL,
finished_at TEXT,
upstream_status INTEGER,
upstream_request_id TEXT NOT NULL DEFAULT '',
context_err TEXT NOT NULL DEFAULT '',
cancellation_source TEXT NOT NULL DEFAULT '',
relay_error TEXT NOT NULL DEFAULT '',
sse_event_count INTEGER NOT NULL DEFAULT 0,
bytes_read INTEGER NOT NULL DEFAULT 0,
content_length INTEGER NOT NULL DEFAULT 0,
last_event TEXT NOT NULL DEFAULT '',
saw_done INTEGER NOT NULL DEFAULT 0
);
CREATE INDEX IF NOT EXISTS request_stream_diagnostics_created_at ON request_stream_diagnostics(created_at DESC);`},
}

const schemaMigrationsDDL = `
Expand Down
137 changes: 114 additions & 23 deletions internal/accounts/request_logs.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -27,29 +27,46 @@ const (
var ErrRequestLogNotFound = errors.New("request log not found")

type RequestLog struct {
ID string `json:"id"`
CreatedAt time.Time `json:"created_at"`
FinishedAt *time.Time `json:"finished_at,omitempty"`
Stream bool `json:"stream"`
Status string `json:"status"`
RequestedModel string `json:"requested_model"`
MappedModel string `json:"mapped_model,omitempty"`
AccountID string `json:"account_id,omitempty"`
Provider string `json:"provider,omitempty"`
Routing string `json:"routing,omitempty"`
PromptTokens *int `json:"prompt_tokens,omitempty"`
CompletionTokens *int `json:"completion_tokens,omitempty"`
CacheReadTokens *int `json:"cache_read_tokens,omitempty"`
CacheWriteTokens *int `json:"cache_write_tokens,omitempty"`
UsageSource string `json:"usage_source,omitempty"`
Credits *float64 `json:"credits,omitempty"`
LatencyMs *int `json:"latency_ms,omitempty"`
TTFBMs *int `json:"ttfb_ms,omitempty"`
ErrorKind string `json:"error_kind,omitempty"`
ErrorCode string `json:"error_code,omitempty"`
ErrorMessage string `json:"error_message,omitempty"`
AttemptCount int `json:"attempt_count"`
Attempts []RequestAttempt `json:"attempts,omitempty"`
ID string `json:"id"`
CreatedAt time.Time `json:"created_at"`
FinishedAt *time.Time `json:"finished_at,omitempty"`
Stream bool `json:"stream"`
Status string `json:"status"`
RequestedModel string `json:"requested_model"`
MappedModel string `json:"mapped_model,omitempty"`
AccountID string `json:"account_id,omitempty"`
Provider string `json:"provider,omitempty"`
Routing string `json:"routing,omitempty"`
PromptTokens *int `json:"prompt_tokens,omitempty"`
CompletionTokens *int `json:"completion_tokens,omitempty"`
CacheReadTokens *int `json:"cache_read_tokens,omitempty"`
CacheWriteTokens *int `json:"cache_write_tokens,omitempty"`
UsageSource string `json:"usage_source,omitempty"`
Credits *float64 `json:"credits,omitempty"`
LatencyMs *int `json:"latency_ms,omitempty"`
TTFBMs *int `json:"ttfb_ms,omitempty"`
ErrorKind string `json:"error_kind,omitempty"`
ErrorCode string `json:"error_code,omitempty"`
ErrorMessage string `json:"error_message,omitempty"`
AttemptCount int `json:"attempt_count"`
Attempts []RequestAttempt `json:"attempts,omitempty"`
StreamDiagnostic *RequestStreamDiagnostic `json:"stream_diagnostic,omitempty"`
}

type RequestStreamDiagnostic struct {
RequestID string `json:"request_id"`
CreatedAt time.Time `json:"created_at"`
FinishedAt *time.Time `json:"finished_at,omitempty"`
UpstreamStatus *int `json:"upstream_status,omitempty"`
UpstreamRequestID string `json:"upstream_request_id,omitempty"`
ContextErr string `json:"context_err,omitempty"`
CancellationSource string `json:"cancellation_source,omitempty"`
RelayError string `json:"relay_error,omitempty"`
SSEEventCount int `json:"sse_event_count"`
BytesRead int64 `json:"bytes_read"`
ContentLength int `json:"content_length"`
LastEvent string `json:"last_event,omitempty"`
SawDone bool `json:"saw_done"`
}

type RequestAttempt struct {
Expand Down Expand Up @@ -634,9 +651,83 @@ func (s *Store) GetRequestLog(ctx context.Context, id string) (RequestLog, error
return RequestLog{}, err
}
log.Attempts = attempts
diagnostic, err := s.getRequestStreamDiagnostic(ctx, log.ID)
if err != nil {
return RequestLog{}, err
}
log.StreamDiagnostic = diagnostic
return log, nil
}

func (s *Store) InsertRequestStreamDiagnostic(ctx context.Context, diagnostic RequestStreamDiagnostic) error {
if strings.TrimSpace(diagnostic.RequestID) == "" {
return fmt.Errorf("request stream diagnostic request id required")
}
if diagnostic.CreatedAt.IsZero() {
diagnostic.CreatedAt = time.Now().UTC()
}
var finished any
if diagnostic.FinishedAt != nil {
finished = formatTime(*diagnostic.FinishedAt)
}
_, err := s.db.ExecContext(ctx, `
INSERT INTO request_stream_diagnostics (
request_id, created_at, finished_at, upstream_status, upstream_request_id, context_err,
cancellation_source, relay_error, sse_event_count, bytes_read, content_length,
last_event, saw_done
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
ON CONFLICT(request_id) DO UPDATE SET
finished_at = excluded.finished_at, upstream_status = excluded.upstream_status,
upstream_request_id = excluded.upstream_request_id, context_err = excluded.context_err,
cancellation_source = excluded.cancellation_source, relay_error = excluded.relay_error,
sse_event_count = excluded.sse_event_count, bytes_read = excluded.bytes_read,
content_length = excluded.content_length, last_event = excluded.last_event,
saw_done = excluded.saw_done`,
diagnostic.RequestID, formatTime(diagnostic.CreatedAt), finished, nullableInt(diagnostic.UpstreamStatus),
diagnostic.UpstreamRequestID, diagnostic.ContextErr, diagnostic.CancellationSource, diagnostic.RelayError,
diagnostic.SSEEventCount, diagnostic.BytesRead, diagnostic.ContentLength, diagnostic.LastEvent,
boolToInt(diagnostic.SawDone),
)
if err != nil {
return fmt.Errorf("insert request stream diagnostic: %w", err)
}
return nil
}

func (s *Store) getRequestStreamDiagnostic(ctx context.Context, requestID string) (*RequestStreamDiagnostic, error) {
var diagnostic RequestStreamDiagnostic
var created, finished, upstreamRequestID, contextErr, cancellationSource, relayError string
var upstreamStatus sql.NullInt64
var sawDone int
err := s.db.QueryRowContext(ctx, `
SELECT request_id, created_at, finished_at, upstream_status, upstream_request_id, context_err,
cancellation_source, relay_error, sse_event_count, bytes_read, content_length,
last_event, saw_done
FROM request_stream_diagnostics WHERE request_id = ?`, requestID).Scan(
&diagnostic.RequestID, &created, &finished, &upstreamStatus, &upstreamRequestID, &contextErr,
&cancellationSource, &relayError, &diagnostic.SSEEventCount, &diagnostic.BytesRead,
&diagnostic.ContentLength, &diagnostic.LastEvent, &sawDone,
)
if errors.Is(err, sql.ErrNoRows) {
return nil, nil
}
if err != nil {
return nil, fmt.Errorf("get request stream diagnostic: %w", err)
}
diagnostic.CreatedAt = parseTime(created)
if finished != "" {
parsed := parseTime(finished)
diagnostic.FinishedAt = &parsed
}
if upstreamStatus.Valid {
diagnostic.UpstreamStatus = nullIntPtr(upstreamStatus)
}
diagnostic.UpstreamRequestID, diagnostic.ContextErr = upstreamRequestID, contextErr
diagnostic.CancellationSource, diagnostic.RelayError = cancellationSource, relayError
diagnostic.SawDone = sawDone != 0
return &diagnostic, nil
}

func (s *Store) listRequestAttempts(ctx context.Context, requestID string) ([]RequestAttempt, error) {
rows, err := s.db.QueryContext(ctx, `
SELECT id, request_id, attempt_index, account_id, started_at, finished_at, status, http_status,
Expand Down
13 changes: 13 additions & 0 deletions internal/accounts/request_logs_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -47,6 +47,15 @@ func TestRequestLogsInsertListGetAndPurge(t *testing.T) {
}); err != nil {
t.Fatal(err)
}
upstreamStatus := 200
if err := store.InsertRequestStreamDiagnostic(ctx, RequestStreamDiagnostic{
RequestID: parentID, CreatedAt: now, FinishedAt: &finished, UpstreamStatus: &upstreamStatus,
ContextErr: "context canceled", CancellationSource: "request_context_canceled",
RelayError: "stream read error: context canceled", SSEEventCount: 4, BytesRead: 512,
ContentLength: -1, LastEvent: "message", SawDone: false,
}); err != nil {
t.Fatal(err)
}
if err := store.InsertRequestAttempt(ctx, RequestAttempt{
ID: NewAttemptID(), RequestID: parentID, AttemptIndex: 1, AccountID: "acc_b",
StartedAt: now, FinishedAt: &finished, Status: AttemptStatusOK,
Expand All @@ -70,6 +79,10 @@ func TestRequestLogsInsertListGetAndPurge(t *testing.T) {
if got.AccountID != wb.ID || got.Provider != "workbuddy" || len(got.Attempts) != 2 || got.Attempts[0].Status != AttemptStatusFailover {
t.Fatalf("detail = %+v", got)
}
if got.StreamDiagnostic == nil || got.StreamDiagnostic.UpstreamStatus == nil || *got.StreamDiagnostic.UpstreamStatus != 200 ||
got.StreamDiagnostic.CancellationSource != "request_context_canceled" || got.StreamDiagnostic.SSEEventCount != 4 || got.StreamDiagnostic.SawDone {
t.Fatalf("stream diagnostic = %+v", got.StreamDiagnostic)
}

filtered, err := store.ListRequestLogs(ctx, RequestLogFilter{AccountID: wb.ID, Status: RequestStatusOK})
if err != nil || filtered.Total != 1 {
Expand Down
Loading
Loading