import { createLogger } from '@shared/utils/logger'; import * as fs from 'fs/promises'; import { TeamMemberLogsFinder } from './TeamMemberLogsFinder'; import type { MemberLogSummary, MemberRuntimeAdvisory, ResolvedTeamMember } from '@shared/types'; interface RuntimeAdvisoryLogFileRef { memberName: string; filePath: string; mtimeMs: number; } interface RuntimeAdvisoryLogsFinder { findMemberLogs( teamName: string, memberName: string, mtimeSinceMs?: number | null ): Promise[]>; findRecentMemberLogFileRefsByMember?( teamName: string, memberNames: readonly string[], mtimeSinceMs?: number | null ): Promise; } const LOOKBACK_MS = 10 * 60 * 1000; const CACHE_TTL_MS = 30_000; const TAIL_BYTES = 64 * 1024; const BATCH_WARN_MS = 1_000; const ADVISORY_FETCH_CONCURRENCY = 2; const QUOTA_EXHAUSTED_TOKENS = [ 'exhausted your capacity', 'capacity exceeded', 'quota exceeded', 'quota exhausted', ]; const RATE_LIMITED_TOKENS = [ 'rate limit', 'too many requests', '429', 'model cooldown', 'cooling down', ]; const AUTH_ERROR_TOKENS = [ 'auth_unavailable', 'no auth available', 'authentication_failed', 'unauthorized', 'forbidden', 'invalid api key', 'authentication', 'api key', 'does not have access', 'please run /login', ]; const CODEX_NATIVE_TIMEOUT_TOKENS = ['codex native exec timed out']; const NETWORK_ERROR_TOKENS = [ 'timeout', 'timed out', 'network', 'connection', 'econn', 'enotfound', 'fetch failed', ]; const PROVIDER_OVERLOADED_TOKENS = [ 'overloaded', 'temporarily unavailable', 'service unavailable', '503', ]; const logger = createLogger('Service:TeamMemberRuntimeAdvisory'); interface CachedRuntimeAdvisory { value: MemberRuntimeAdvisory | null; expiresAt: number; } interface CachedTeamBatchAdvisories { membersSignature: string; value: Map; expiresAt: number; } function includesAnyToken(value: string, tokens: readonly string[]): boolean { return tokens.some((token) => value.includes(token)); } function classifyRetryReason(message: string | undefined): MemberRuntimeAdvisory['reasonCode'] { const normalized = message?.trim().toLowerCase(); if (!normalized) { return 'unknown'; } if (includesAnyToken(normalized, QUOTA_EXHAUSTED_TOKENS)) { return 'quota_exhausted'; } if (includesAnyToken(normalized, RATE_LIMITED_TOKENS)) { return 'rate_limited'; } if (includesAnyToken(normalized, AUTH_ERROR_TOKENS)) { return 'auth_error'; } if (includesAnyToken(normalized, CODEX_NATIVE_TIMEOUT_TOKENS)) { return 'codex_native_timeout'; } if (includesAnyToken(normalized, NETWORK_ERROR_TOKENS)) { return 'network_error'; } if (includesAnyToken(normalized, PROVIDER_OVERLOADED_TOKENS)) { return 'provider_overloaded'; } return 'backend_error'; } async function mapLimit( items: readonly T[], limit: number, fn: (item: T) => Promise ): Promise { const results = new Array(items.length); let index = 0; const workerCount = Math.max(1, Math.min(limit, items.length)); const workers = new Array(workerCount).fill(0).map(async () => { while (true) { const currentIndex = index; index += 1; if (currentIndex >= items.length) { return; } results[currentIndex] = await fn(items[currentIndex]); } }); await Promise.all(workers); return results; } export class TeamMemberRuntimeAdvisoryService { private readonly memberCache = new Map(); private readonly teamBatchCacheByTeam = new Map(); private readonly inFlightBatchRequests = new Map< string, Promise> >(); constructor( private readonly logsFinder: RuntimeAdvisoryLogsFinder = new TeamMemberLogsFinder() ) {} async getMemberAdvisories( teamName: string, members: readonly Pick[] ): Promise> { const activeMembers = members.filter((member) => !member.removedAt); if (activeMembers.length === 0) { return new Map(); } const teamKey = this.normalizeToken(teamName); const membersSignature = this.buildMembersSignature(activeMembers); const now = Date.now(); const cachedBatch = this.teamBatchCacheByTeam.get(teamKey); if (cachedBatch?.membersSignature === membersSignature && cachedBatch.expiresAt > now) { return this.materializeBatchAdvisories(activeMembers, cachedBatch.value); } const inFlightKey = `${teamKey}::${membersSignature}`; const existingRequest = this.inFlightBatchRequests.get(inFlightKey); if (existingRequest) { return this.materializeBatchAdvisories(activeMembers, await existingRequest); } const request = this.loadBatchAdvisories(teamName, teamKey, activeMembers, membersSignature); this.inFlightBatchRequests.set(inFlightKey, request); try { return this.materializeBatchAdvisories(activeMembers, await request); } finally { if (this.inFlightBatchRequests.get(inFlightKey) === request) { this.inFlightBatchRequests.delete(inFlightKey); } } } async getMemberAdvisory( teamName: string, memberName: string ): Promise { const cacheKey = this.getMemberCacheKey(teamName, memberName); const cached = this.memberCache.get(cacheKey); if (cached && cached.expiresAt > Date.now()) { return cached.value ? this.cloneAdvisory(cached.value) : null; } const advisory = await this.findRecentMemberAdvisory(teamName, memberName); this.memberCache.set(cacheKey, { value: advisory, expiresAt: Date.now() + CACHE_TTL_MS, }); return advisory ? this.cloneAdvisory(advisory) : null; } private async loadBatchAdvisories( teamName: string, teamKey: string, activeMembers: readonly Pick[], membersSignature: string ): Promise> { const startedAt = performance.now(); const now = Date.now(); const result = new Map(); const membersToFetch: string[] = []; let memberCacheHits = 0; let memberCacheMisses = 0; for (const member of activeMembers) { const normalizedMemberName = this.normalizeToken(member.name); const cacheKey = `${teamKey}::${normalizedMemberName}`; const cached = this.memberCache.get(cacheKey); if (cached && cached.expiresAt > now) { memberCacheHits += 1; if (cached.value) { result.set(normalizedMemberName, this.cloneAdvisory(cached.value)); } continue; } memberCacheMisses += 1; membersToFetch.push(member.name); } if (membersToFetch.length > 0) { const fetched = await this.findRecentMemberAdvisories(teamName, membersToFetch); const fetchedAt = Date.now(); for (const [memberName, advisory] of fetched) { const normalizedMemberName = this.normalizeToken(memberName); this.memberCache.set(`${teamKey}::${normalizedMemberName}`, { value: advisory, expiresAt: fetchedAt + CACHE_TTL_MS, }); if (advisory) { result.set(normalizedMemberName, this.cloneAdvisory(advisory)); } } } this.teamBatchCacheByTeam.set(teamKey, { membersSignature, value: this.cloneNormalizedAdvisories(result), expiresAt: Date.now() + CACHE_TTL_MS, }); const totalMs = performance.now() - startedAt; if (totalMs >= BATCH_WARN_MS) { logger.warn( `[perf] getMemberAdvisories slow team=${teamName} activeMembers=${activeMembers.length} signatureMembers=${activeMembers.length} batchCache=miss memberCacheHits=${memberCacheHits} memberCacheMisses=${memberCacheMisses} fetchedMembers=${membersToFetch.length} total=${totalMs.toFixed(1)}ms` ); } return result; } private getMemberCacheKey(teamName: string, memberName: string): string { return `${this.normalizeToken(teamName)}::${this.normalizeToken(memberName)}`; } private buildMembersSignature(members: readonly Pick[]): string { return Array.from(new Set(members.map((member) => this.normalizeToken(member.name)))) .sort() .join('|'); } private normalizeToken(value: string): string { return value.trim().toLowerCase(); } private cloneAdvisory(advisory: MemberRuntimeAdvisory): MemberRuntimeAdvisory { return { ...advisory }; } private cloneNormalizedAdvisories( advisories: ReadonlyMap ): Map { return new Map( Array.from(advisories, ([memberName, advisory]) => [memberName, this.cloneAdvisory(advisory)]) ); } private materializeBatchAdvisories( activeMembers: readonly Pick[], advisories: ReadonlyMap ): Map { const materialized = new Map(); for (const member of activeMembers) { const advisory = advisories.get(this.normalizeToken(member.name)); if (advisory) { materialized.set(member.name, this.cloneAdvisory(advisory)); } } return materialized; } private async findRecentMemberAdvisory( teamName: string, memberName: string ): Promise { const summaries = await this.logsFinder.findMemberLogs( teamName, memberName, Date.now() - LOOKBACK_MS ); return this.findRecentMemberAdvisoryInFiles( summaries.flatMap((summary) => summary.filePath ?? []) ); } private async findRecentMemberAdvisories( teamName: string, memberNames: readonly string[] ): Promise { if (this.logsFinder.findRecentMemberLogFileRefsByMember) { try { return await this.findRecentMemberAdvisoriesFromBatchRefs(teamName, memberNames); } catch (error) { logger.warn('batch member runtime advisory log lookup failed; falling back', { teamName, error: error instanceof Error ? error.message : String(error), }); } } return mapLimit(memberNames, ADVISORY_FETCH_CONCURRENCY, async (memberName) => { const advisory = await this.findRecentMemberAdvisory(teamName, memberName); return [memberName, advisory] as const; }); } private async findRecentMemberAdvisoriesFromBatchRefs( teamName: string, memberNames: readonly string[] ): Promise { const memberNamesByKey = new Map(); for (const memberName of memberNames) { const normalized = this.normalizeToken(memberName); if (!memberNamesByKey.has(normalized)) { memberNamesByKey.set(normalized, memberName); } } const refs = await this.logsFinder.findRecentMemberLogFileRefsByMember!( teamName, memberNames, Date.now() - LOOKBACK_MS ); const refsByMember = new Map(); for (const ref of refs) { const normalizedMemberName = this.normalizeToken(ref.memberName); if (!memberNamesByKey.has(normalizedMemberName)) { continue; } const bucket = refsByMember.get(normalizedMemberName) ?? []; bucket.push(ref); refsByMember.set(normalizedMemberName, bucket); } return mapLimit(memberNames, ADVISORY_FETCH_CONCURRENCY, async (memberName) => { const normalizedMemberName = this.normalizeToken(memberName); const refsForMember = refsByMember.get(normalizedMemberName) ?? []; const seenFilePaths = new Set(); const filePaths = refsForMember .slice() .sort((left, right) => right.mtimeMs - left.mtimeMs) .flatMap((ref) => { if (!ref.filePath || seenFilePaths.has(ref.filePath)) { return []; } seenFilePaths.add(ref.filePath); return [ref.filePath]; }); return [memberName, await this.findRecentMemberAdvisoryInFiles(filePaths)] as const; }); } private async findRecentMemberAdvisoryInFiles( filePaths: readonly string[] ): Promise { for (const filePath of filePaths) { const advisory = await this.readRecentApiRetryAdvisory(filePath); if (advisory) { return advisory; } } return null; } private async readRecentApiRetryAdvisory( filePath: string ): Promise { let handle: fs.FileHandle | null = null; try { handle = await fs.open(filePath, 'r'); const stat = await handle.stat(); if (!stat.isFile() || stat.size <= 0) { return null; } const start = Math.max(0, stat.size - TAIL_BYTES); const buffer = Buffer.alloc(stat.size - start); if (buffer.length === 0) { return null; } await handle.read(buffer, 0, buffer.length, start); const tail = buffer.toString('utf8'); const lines = tail.split('\n'); if (start > 0) { lines.shift(); } const now = Date.now(); for (let index = lines.length - 1; index >= 0; index -= 1) { const line = lines[index]?.trim() ?? ''; const advisory = this.extractApiRetryAdvisory(line, now) ?? this.extractApiErrorAdvisory(line, now); if (advisory) { return advisory; } } return null; } catch { return null; } finally { await handle?.close().catch(() => {}); } } private extractApiRetryAdvisory(line: string, now = Date.now()): MemberRuntimeAdvisory | null { if ( !line || (!line.includes('"subtype":"api_error"') && !line.includes('"subtype": "api_error"')) ) { return null; } try { const parsed = JSON.parse(line) as { type?: string; subtype?: string; retryInMs?: number; timestamp?: string; error?: { message?: string; error?: { message?: string; error?: { message?: string; }; }; }; }; if (parsed.type !== 'system' || parsed.subtype !== 'api_error') { return null; } const retryInMs = typeof parsed.retryInMs === 'number' && Number.isFinite(parsed.retryInMs) && parsed.retryInMs > 0 ? parsed.retryInMs : null; const observedAt = typeof parsed.timestamp === 'string' ? Date.parse(parsed.timestamp) : Number.NaN; if (!retryInMs || !Number.isFinite(observedAt)) { return null; } const retryUntil = observedAt + retryInMs; if (retryUntil <= now) { return null; } const message = parsed.error?.error?.error?.message?.trim() || parsed.error?.error?.message?.trim() || parsed.error?.message?.trim() || undefined; return { kind: 'sdk_retrying', observedAt: new Date(observedAt).toISOString(), retryUntil: new Date(retryUntil).toISOString(), retryDelayMs: retryInMs, reasonCode: classifyRetryReason(message), ...(message ? { message } : {}), }; } catch { return null; } } private extractApiErrorAdvisory(line: string, now = Date.now()): MemberRuntimeAdvisory | null { if ( !line || (!line.includes('"isApiErrorMessage":true') && !line.includes('"isApiErrorMessage": true') && !line.includes('"error":"authentication_failed"') && !line.includes('"error": "authentication_failed"')) ) { return null; } try { const parsed = JSON.parse(line) as { type?: string; timestamp?: string; error?: string; isApiErrorMessage?: boolean; message?: { content?: { type?: string; text?: string }[]; }; }; if (parsed.type !== 'assistant') { return null; } const observedAt = typeof parsed.timestamp === 'string' ? Date.parse(parsed.timestamp) : Number.NaN; if (!Number.isFinite(observedAt) || observedAt < now - LOOKBACK_MS) { return null; } const message = this.extractAssistantText(parsed.message?.content); if (!parsed.isApiErrorMessage && parsed.error !== 'authentication_failed') { return null; } if (!message && parsed.error !== 'authentication_failed') { return null; } const statusMatch = /^API Error:\s*(\d{3})/.exec(message); return { kind: 'api_error', observedAt: new Date(observedAt).toISOString(), reasonCode: classifyRetryReason(message || parsed.error), ...(message ? { message } : {}), ...(statusMatch ? { statusCode: Number(statusMatch[1]) } : {}), }; } catch { return null; } } private extractAssistantText(content: { type?: string; text?: string }[] | undefined): string { if (!Array.isArray(content)) { return ''; } return content .filter((item) => item.type === 'text' && typeof item.text === 'string') .map((item) => item.text?.trim()) .filter(Boolean) .join('\n') .trim(); } }