Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
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
9 changes: 9 additions & 0 deletions changelog/unreleased/live-selecting-queue.md
Original file line number Diff line number Diff line change
@@ -0,0 +1,9 @@
### English

- Overview InflightLiveCard waiting rail now shows real selecting requests from `/api/live` `queue` (Prepare → pool Acquire), reusing existing `request_logs` statuses (`started` / `streaming` / terminal) instead of a parallel inflight API.
- Worker rail order is busy → available → cooldown, and rate-limit remaining time formats with hours when the window is long enough (`Nh Nm`).

### 中文

- Overview「实时进行中」等待队列改为读取 `/api/live` 的 `queue`(Prepare → 选号 → Acquire),复用现有 `request_logs` 状态(`started` / `streaming` / 终态),不另开 `/api/inflight*`。
- 账号槽排序为忙碌 → 空闲 → 冷却,限流剩余时间在足够长时按小时显示(`Nh Nm`)。
47 changes: 47 additions & 0 deletions frontend/scripts/mockInflight.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -3,6 +3,7 @@ import {
applyLiveProjection,
cloneMockInflight,
getMockInflightSnapshot,
liveHasSignal,
MOCK_INFLIGHT,
} from '../src/lib/mockInflight.ts'

Expand Down Expand Up @@ -281,3 +282,49 @@ run('applyLiveProjection maps max_inflight / in_flight to slot cells', () => {
})

console.log('all mockInflight tests passed')

run('applyLiveProjection maps queue onto queued MockJobStatus', () => {
const started = new Date(Date.now() - 5_000).toISOString()
const snap = applyLiveProjection(getMockInflightSnapshot(), {
in_flight: 1,
rpm: 2,
workers: [
{
id: 'acc-1',
name: 'wb-a1',
provider: 'workbuddy',
busy: true,
progress: 0.4,
rate_limited: false,
},
],
rate_limits: [],
sticky: { bindings: 0, hits: 0, misses: 0, capacity: 64, ttl_seconds: 600 },
queue: [
{ id: 'req-1', name: 'chat · wait', status: 'started', started_at: started },
{ id: 'req-2', name: 'embed · docs', status: 'started' },
],
})
const queued = snap.jobs.filter((j) => j.status === 'queued')
const running = snap.jobs.filter((j) => j.status === 'running')
assert.equal(running.length, 1)
assert.equal(queued.length, 2)
assert.equal(queued[0].id, 'req-1')
assert.equal(queued[0].name, 'chat · wait')
assert.ok(queued[0].elapsedMs >= 4_000 && queued[0].elapsedMs < 15_000)
assert.equal(queued[1].id, 'req-2')
})

run('liveHasSignal true when only queue present', () => {
assert.equal(
liveHasSignal({
in_flight: 0,
rpm: 0,
workers: [],
rate_limits: [],
sticky: { bindings: 0, hits: 0, misses: 0, capacity: 0, ttl_seconds: 0 },
queue: [{ id: 'r1', name: 'm', status: 'started' }],
}),
true,
)
})
15 changes: 15 additions & 0 deletions frontend/src/api/live.ts
Original file line number Diff line number Diff line change
Expand Up @@ -26,6 +26,19 @@ export type LiveCompletion = {
status: string
}

/**
* Selecting (选号) request for the InflightLiveCard waiting rail.
* status reuses request_logs RequestStatus* (typically started) — not a new machine.
*/
export type LiveJob = {
id: string
name: string
/** RequestStatus* from recorder (started while selecting). */
status: string
/** RFC3339 Prepare/Start time. */
started_at?: string
}

/** Active cooldown window from AccountView.DownUntil / model_cooldowns. */
export type LiveRateLimit = {
id: string
Expand Down Expand Up @@ -62,6 +75,8 @@ export type LiveSnapshot = {
sticky: LiveSticky
/** Terminal finishes since last poll — InflightLiveCard routes fly-outs. */
recent_completions?: LiveCompletion[]
/** Prepare→Acquire 选号 (RequestStatusStarted, no pool bind). */
queue?: LiveJob[]
}

export function fetchLive() {
Expand Down
13 changes: 8 additions & 5 deletions frontend/src/components/overview/InflightLiveCard.tsx
Original file line number Diff line number Diff line change
Expand Up @@ -43,8 +43,10 @@ type T = (key: string, vars?: Record<string, string | number>) => string

function formatRemain(ms: number) {
const sec = Math.max(0, Math.round(ms / 1000))
const m = Math.floor(sec / 60)
const h = Math.floor(sec / 3600)
const m = Math.floor((sec % 3600) / 60)
const s = sec % 60
if (h) return `${h}h ${m}m`
if (m) return `${m}m ${String(s).padStart(2, '0')}s`
return `${s}s`
}
Expand Down Expand Up @@ -401,13 +403,14 @@ export function InflightLiveCard({
const waiting = snap.jobs.filter((j) => j.status === 'queued')
const queuePills = waiting.slice(0, QUEUE_PILL_MAX)
const queueOverflow = waiting.slice(QUEUE_PILL_MAX)
// AffectUI: short worker rail — busy / rate-limited first, cap at N (not every account).
// AffectUI: short worker rail — busy → available → cooldown, cap at N.
const visibleWorkers = useMemo(() => {
const preferred = snap.workers.filter((w) => w.busy || w.rateLimited)
const rest = snap.workers.filter((w) => !w.busy && !w.rateLimited)
const busy = snap.workers.filter((w) => w.busy)
const available = snap.workers.filter((w) => !w.busy && !w.rateLimited)
const cooling = snap.workers.filter((w) => !w.busy && w.rateLimited)
const out: typeof snap.workers = []
const seen = new Set<string>()
for (const w of [...preferred, ...rest]) {
for (const w of [...busy, ...available, ...cooling]) {
if (seen.has(w.id)) continue
seen.add(w.id)
out.push(w)
Expand Down
50 changes: 46 additions & 4 deletions frontend/src/lib/mockInflight.ts
Original file line number Diff line number Diff line change
Expand Up @@ -6,9 +6,9 @@
* rateLimits≈ cooldown / provider rate-limit windows
* depth ≈ recent waiting + concurrency approx
* totals ≈ recent request_logs Done/Failed (mock until Phase 2)
* Stage-3 swaps workers/rateLimits/sticky onto /api/live (LiveWorker,
* LiveRateLimit, LiveSticky); keep jobs/depth mock-only until request-level
* inflight lands. Do not invent a parallel /api/inflight*.
* Stage-3 swaps workers/rateLimits/sticky/queue onto /api/live (LiveWorker,
* LiveRateLimit, LiveSticky, LiveJob); depth stays client-rolled. Do not invent
* a parallel /api/inflight*.
*
* Live loop (`tickMockInflight`) mirrors public AffectUI Motion prose:
* progress fill, queue→worker handoff, DONE/FAILED stack, depth scroll —
Expand Down Expand Up @@ -409,13 +409,23 @@ export type LiveStickyLike = {
escapes?: number
}

export type LiveQueueJobLike = {
id: string
name: string
/** RequestStatus* — UI maps selecting onto MockJobStatus queued. */
status: string
started_at?: string
}

export type LiveProjectionInput = {
in_flight: number
rpm: number
workers: LiveWorkerLike[]
rate_limits: LiveRateLimitLike[]
sticky: LiveStickyLike
recent_completions?: LiveCompletionLike[]
/** Selecting requests (started, unbound) from /api/live queue. */
queue?: LiveQueueJobLike[]
}

/**
Expand Down Expand Up @@ -471,7 +481,7 @@ export function applyLiveProjection(
}
})

const jobs: MockInflightJob[] = workers
const runningJobs: MockInflightJob[] = workers
.filter((w) => w.busy && w.jobId)
.map((w) => {
const id = w.jobId!
Expand Down Expand Up @@ -501,6 +511,37 @@ export function applyLiveProjection(
}
})

// Selecting queue: RequestStatusStarted (etc.) → UI MockJobStatus queued.
// Do not invent parallel statuses — only map existing request_logs values.
const queuedJobs: MockInflightJob[] = (live.queue ?? []).map((q) => {
const id = (q.id || '').trim()
const prev = prevById.get(id)
let startedAtMs: number | undefined
const raw = q.started_at?.trim()
if (raw) {
const parsed = Date.parse(raw)
if (!Number.isNaN(parsed)) startedAtMs = parsed
}
if (startedAtMs == null && prev?.startedAtMs != null) {
startedAtMs = prev.startedAtMs
}
if (startedAtMs == null) {
startedAtMs = now
}
const name = (q.name || '').trim() || id
return {
id,
name,
kind: 'chat' as const,
status: 'queued' as const,
progress: 0,
startedAtMs,
elapsedMs: Math.max(0, now - startedAtMs),
}
})

const jobs: MockInflightJob[] = [...runningJobs, ...queuedJobs]

// Continue rolling depth only after a prior live projection (cleared p95 +
// zero totals). Mock seed/tick always leaves waitP95Ms > 0 — start zeros.
const priorIsLive = base.waitP95Ms === 0 && base.totals.done === 0 && base.totals.failed === 0
Expand Down Expand Up @@ -565,6 +606,7 @@ export function liveHasSignal(live: LiveProjectionInput): boolean {
live.workers.length > 0 ||
live.rate_limits.length > 0 ||
live.in_flight > 0 ||
(live.queue?.length ?? 0) > 0 ||
(live.sticky?.bindings ?? 0) > 0
)
}
2 changes: 2 additions & 0 deletions internal/app/app.go
Original file line number Diff line number Diff line change
Expand Up @@ -157,6 +157,8 @@ func New(cfg config.Config) *App {
chatExecutor.MaxAttempts = cfg.MaxRetryAccounts
chatExecutor.Providers = providerReg
chatExecutor.OnAttempt = recorder.Attempt
chatExecutor.OnInFlightAcquire = recorder.BindAccount
chatExecutor.OnInFlightRelease = recorder.UnbindAccount
a := &App{
Cfg: cfg,
Auth: auth.NewVerifier(proxyAPIKey, store),
Expand Down
47 changes: 38 additions & 9 deletions internal/console/live.go
Original file line number Diff line number Diff line change
Expand Up @@ -66,16 +66,28 @@ type LiveCompletion struct {
Status string `json:"status"`
}

// LiveJob is one selecting (选号) request for the InflightLiveCard waiting
// queue. Status reuses accounts.RequestStatus* (started) — not a new state
// machine. Running occupancy stays on Workers via Pool InFlight.
type LiveJob struct {
ID string `json:"id"`
Name string `json:"name"`
Status string `json:"status"`
StartedAt string `json:"started_at,omitempty"`
}

type LiveSnapshot struct {
Time string `json:"time"`
InFlight int `json:"in_flight"`
Capacity int `json:"capacity"`
RPM int `json:"rpm"`
Accounts []LiveAccount `json:"accounts"`
Workers []LiveWorker `json:"workers"`
RateLimits []LiveRateLimit `json:"rate_limits"`
Sticky executor.SessionAffinityStats `json:"sticky"`
RecentCompletions []LiveCompletion `json:"recent_completions,omitempty"`
Time string `json:"time"`
InFlight int `json:"in_flight"`
Capacity int `json:"capacity"`
RPM int `json:"rpm"`
Accounts []LiveAccount `json:"accounts"`
Workers []LiveWorker `json:"workers"`
RateLimits []LiveRateLimit `json:"rate_limits"`
Sticky executor.SessionAffinityStats `json:"sticky"`
RecentCompletions []LiveCompletion `json:"recent_completions,omitempty"`
// Queue is Prepare→Acquire 选号 (RequestStatusStarted, no pool bind).
Queue []LiveJob `json:"queue,omitempty"`
}

// buildLiveSnapshot sums in-flight over all accounts and capacity (the
Expand All @@ -89,6 +101,7 @@ func buildLiveSnapshot(views []accounts.AccountView, rpm int, sticky executor.Se
Accounts: make([]LiveAccount, 0, len(views)),
Workers: make([]LiveWorker, 0, len(views)),
RateLimits: make([]LiveRateLimit, 0),
Queue: make([]LiveJob, 0),
Sticky: sticky,
}
for _, v := range views {
Expand Down Expand Up @@ -205,6 +218,22 @@ func (h *Handler) HandleLive(w http.ResponseWriter, r *http.Request) {
Status: c.Status,
})
}
for _, q := range h.Recorder.SelectingRequests() {
name := strings.TrimSpace(q.RequestedModel)
if name == "" {
name = q.ID
}
status := strings.TrimSpace(q.Status)
if status == "" {
status = accounts.RequestStatusStarted
}
snap.Queue = append(snap.Queue, LiveJob{
ID: q.ID,
Name: name,
Status: status,
StartedAt: q.StartedAt.UTC().Format(time.RFC3339),
})
}
}
writeJSON(w, http.StatusOK, snap)
}
6 changes: 6 additions & 0 deletions internal/console/live_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -78,6 +78,12 @@ func TestBuildLiveSnapshot(t *testing.T) {
if got.RateLimits[0].Label != "wb-a2" || got.RateLimits[1].Label != "wb-a2 · glm-5" {
t.Fatalf("labels = %q / %q", got.RateLimits[0].Label, got.RateLimits[1].Label)
}
if got.Queue == nil {
t.Fatalf("queue must be non-nil empty slice")
}
if len(got.Queue) != 0 {
t.Fatalf("queue = %+v", got.Queue)
}
}

func TestBuildLiveSnapshotEmpty(t *testing.T) {
Expand Down
Loading
Loading