mirror of
https://github.com/wu736139669/hapi.git
synced 2026-10-15 20:29:55 +00:00
revert(codex): remove Luna Reserve fallback (#1780)
Revert 8357da0a9d and its later shared-runtime integration.
Remove automatic model fallback, account usage polling, and the related UI, protocol fields, tests, and documentation without adding replacement quota handling.
Keep generic model pagination and method probing required by the current shared-session architecture. Cover idle sessions staying online without usage polling or agent-state churn.
This commit is contained in:
weishu
@@ -387,11 +387,3 @@ export interface ExperimentalFeatureEnablementSetResponse {
|
||||
enablement: Record<string, boolean>;
|
||||
[key: string]: unknown;
|
||||
}
|
||||
|
||||
export interface ThreadSettingsUpdateParams {
|
||||
threadId: string;
|
||||
model: string;
|
||||
effort?: string | null;
|
||||
serviceTier?: string | null;
|
||||
collaborationMode?: CollaborationMode;
|
||||
}
|
||||
|
||||
@@ -62,26 +62,6 @@ describe('CodexAppServerClient process cwd', () => {
|
||||
spawnMock.mockReset();
|
||||
});
|
||||
|
||||
it('uses null for legacy usage reads and opts in only when Reserve is supported', async () => {
|
||||
const child = fakeChild();
|
||||
child.stdin.write = vi.fn((_data: unknown, callback?: (error?: Error | null) => void) => {
|
||||
callback?.();
|
||||
return true;
|
||||
});
|
||||
spawnMock.mockReturnValue(child);
|
||||
const client = new CodexAppServerClient({ cwd: '/neutral-home' });
|
||||
await client.connect();
|
||||
for (const supported of [false, true]) {
|
||||
const pending = client.readAccountRateLimits(supported);
|
||||
const request = JSON.parse(child.stdin.write.mock.lastCall![0] as string);
|
||||
expect(request.method).toBe('account/rateLimits/read');
|
||||
expect(request.params).toEqual(supported ? { supportsLunaReserve: true } : null);
|
||||
child.stdout.emit('data', Buffer.from(JSON.stringify({ id: request.id, result: { rateLimits: {} } }) + '\n'));
|
||||
await expect(pending).resolves.toEqual({ rateLimits: {} });
|
||||
}
|
||||
await client.disconnect();
|
||||
});
|
||||
|
||||
it('passes an explicit neutral cwd to the app-server process', async () => {
|
||||
spawnMock.mockReturnValue(fakeChild());
|
||||
const client = new CodexAppServerClient({ cwd: '/neutral-home' });
|
||||
|
||||
@@ -5,7 +5,6 @@ import { logger } from '@/ui/logger';
|
||||
import { JsonLineParser } from '@/utils/jsonLineParser';
|
||||
import { killProcessByChildProcess } from '@/utils/process';
|
||||
import type {
|
||||
ThreadSettingsUpdateParams,
|
||||
CollaborationModeListResponse,
|
||||
InitializeParams,
|
||||
InitializeResponse,
|
||||
@@ -356,14 +355,6 @@ export class CodexAppServerClient extends JsonLineParser {
|
||||
return response as ConfigReadResponse;
|
||||
}
|
||||
|
||||
async readAccountRateLimits(supportsLunaReserve = false): Promise<unknown> {
|
||||
return this.sendRequest('account/rateLimits/read', supportsLunaReserve ? { supportsLunaReserve: true } : null, { timeoutMs: 30_000 });
|
||||
}
|
||||
|
||||
async updateThreadSettings(params: ThreadSettingsUpdateParams): Promise<void> {
|
||||
await this.sendRequest('thread/settings/update', params, { timeoutMs: 30_000 });
|
||||
}
|
||||
|
||||
async listModels(params?: ModelListParams): Promise<ModelListResponse> {
|
||||
const response = await this.sendRequest('model/list', params ?? {}, {
|
||||
timeoutMs: 30_000
|
||||
|
||||
@@ -48,7 +48,7 @@ function extractTurnContextModel(event: CodexSessionEvent): string | null | unde
|
||||
const model = (event.payload as Record<string, unknown>).model;
|
||||
if (model === null) return null;
|
||||
if (typeof model !== 'string' || !model.trim()) return undefined;
|
||||
return model.trim() === 'gpt-reserve' ? 'gpt-5.6-luna' : model.trim();
|
||||
return model.trim();
|
||||
}
|
||||
|
||||
export async function codexLocalLauncher(session: CodexSession): Promise<'switch' | 'exit'> {
|
||||
@@ -445,7 +445,7 @@ export async function codexLocalLauncher(session: CodexSession): Promise<'switch
|
||||
await codexLocal({
|
||||
path: session.path,
|
||||
sessionId: resumeSessionId,
|
||||
modelReasoningEffort: resumeSessionId ? undefined : (session.getModelReasoningEffort() ?? undefined) as ReasoningEffort | undefined,
|
||||
modelReasoningEffort: (session.getModelReasoningEffort() ?? undefined) as ReasoningEffort | undefined,
|
||||
onSessionFound: handleSessionFound,
|
||||
abort: abortSignal,
|
||||
codexArgs,
|
||||
|
||||
@@ -1,16 +1,8 @@
|
||||
import { afterEach, describe, expect, it, vi } from 'vitest';
|
||||
import { mkdtemp, rm } from 'node:fs/promises';
|
||||
import { tmpdir } from 'node:os';
|
||||
import { join } from 'node:path';
|
||||
import { MessageQueue2 } from '@/utils/MessageQueue2';
|
||||
import type { EnhancedMode } from './loop';
|
||||
|
||||
const harness = vi.hoisted(() => ({
|
||||
appServerConnected: true,
|
||||
abandonTransport: null as (() => void) | null,
|
||||
resumeModel: 'gpt-5.4',
|
||||
reserveUsage: null as Record<string, unknown> | null,
|
||||
reserveSettingsCalls: [] as Array<Record<string, unknown>>,
|
||||
notifications: [] as Array<{ method: string; params: unknown }>,
|
||||
dispatchNotification: null as ((method: string, params: unknown) => void) | null,
|
||||
registerRequestCalls: [] as string[],
|
||||
@@ -112,14 +104,14 @@ vi.mock('./codexAppServerClient', () => {
|
||||
private notificationHandler: ((method: string, params: unknown) => void) | null = null;
|
||||
private stderrHandler: ((text: string) => void) | null = null;
|
||||
|
||||
async connect(): Promise<void> { harness.appServerConnected = true; }
|
||||
async connect(): Promise<void> {}
|
||||
|
||||
isConnected(): boolean {
|
||||
return harness.appServerConnected;
|
||||
return true;
|
||||
}
|
||||
|
||||
isInitialized(): boolean {
|
||||
return harness.appServerConnected;
|
||||
return true;
|
||||
}
|
||||
|
||||
async initialize(params: unknown): Promise<{ protocolVersion: number }> {
|
||||
@@ -140,7 +132,7 @@ vi.mock('./codexAppServerClient', () => {
|
||||
harness.dispatchNotification = handler;
|
||||
}
|
||||
|
||||
setTransportAbandonedHandler(handler: (() => void) | null): void { harness.abandonTransport = handler; }
|
||||
setTransportAbandonedHandler(_handler: (() => void) | null): void {}
|
||||
|
||||
setStderrHandler(handler: ((text: string) => void) | null): void {
|
||||
this.stderrHandler = handler;
|
||||
@@ -154,25 +146,6 @@ vi.mock('./codexAppServerClient', () => {
|
||||
return harness.collaborationModeResponse;
|
||||
}
|
||||
|
||||
async readAccountRateLimits(): Promise<unknown> {
|
||||
if (!harness.reserveUsage) throw new Error('Unsupported');
|
||||
return structuredClone(harness.reserveUsage);
|
||||
}
|
||||
|
||||
async supportsMethod(): Promise<boolean> { return true; }
|
||||
|
||||
async listModels(): Promise<unknown> {
|
||||
return { data: [
|
||||
{ id: 'gpt-reserve', hidden: true, defaultReasoningEffort: 'medium' },
|
||||
{ id: 'gpt-5.4', hidden: false, defaultReasoningEffort: 'high' }
|
||||
] };
|
||||
}
|
||||
|
||||
async updateThreadSettings(params: Record<string, unknown>): Promise<void> {
|
||||
harness.reserveSettingsCalls.push(params);
|
||||
this.notificationHandler?.('thread/settings/updated', { threadId: params.threadId, threadSettings: params });
|
||||
}
|
||||
|
||||
async listSkills(params: unknown): Promise<unknown> {
|
||||
harness.listSkillsCalls.push(params);
|
||||
return harness.skillsListResponse;
|
||||
@@ -205,7 +178,7 @@ vi.mock('./codexAppServerClient', () => {
|
||||
if (harness.failResumeThreadIds.includes(id)) {
|
||||
throw new Error('resume failed');
|
||||
}
|
||||
return { thread: { id }, model: harness.resumeModel };
|
||||
return { thread: { id }, model: 'gpt-5.4' };
|
||||
}
|
||||
|
||||
async compactThread(params?: { threadId?: string }): Promise<Record<string, never>> {
|
||||
@@ -1136,7 +1109,6 @@ import { INDETERMINATE_SYMBOL } from './codexAppServerClient';
|
||||
import { RPC_METHODS } from '@hapi/protocol/rpcMethods';
|
||||
|
||||
type FakeAgentState = {
|
||||
codexUsage?: import('@hapi/protocol').AgentState['codexUsage'];
|
||||
requests: Record<string, unknown>;
|
||||
completedRequests: Record<string, unknown>;
|
||||
};
|
||||
@@ -1232,10 +1204,6 @@ function createSessionStub(
|
||||
setModelReasoningEffort(nextEffort: EnhancedMode['modelReasoningEffort']) {
|
||||
currentModelReasoningEffort = nextEffort;
|
||||
},
|
||||
getModelReasoningEffort() { return currentModelReasoningEffort; },
|
||||
pushKeepAlive() {},
|
||||
getServiceTier() { return undefined; },
|
||||
setServiceTier() {},
|
||||
getCollaborationMode() {
|
||||
return currentCollaborationMode;
|
||||
},
|
||||
@@ -1431,11 +1399,6 @@ describe('codexRemoteLauncher', () => {
|
||||
session.queue.close();
|
||||
});
|
||||
afterEach(() => {
|
||||
harness.appServerConnected = true;
|
||||
harness.abandonTransport = null;
|
||||
harness.resumeModel = 'gpt-5.4';
|
||||
harness.reserveUsage = null;
|
||||
harness.reserveSettingsCalls = [];
|
||||
harness.notifications = [];
|
||||
harness.dispatchNotification = null;
|
||||
harness.registerRequestCalls = [];
|
||||
@@ -2695,118 +2658,6 @@ describe('codexRemoteLauncher', () => {
|
||||
expect(session.thinking).toBe(false);
|
||||
});
|
||||
|
||||
it('preserves queued model settings when resuming an ordinary thread', async () => {
|
||||
const { session } = createSessionStub(['first'], {
|
||||
...createMode(), model: 'gpt-5.6', modelReasoningEffort: 'high', serviceTier: 'fast'
|
||||
}, true);
|
||||
session.sessionId = 'thread-old';
|
||||
await codexRemoteLauncher(session as never);
|
||||
expect(harness.startTurnParams[0]).toMatchObject({
|
||||
collaborationMode: { settings: { model: 'gpt-5.6', reasoning_effort: 'high' } },
|
||||
serviceTier: 'priority'
|
||||
});
|
||||
});
|
||||
|
||||
it.each([
|
||||
['transient', 'Codex thread entered systemError', 2, false],
|
||||
['context', "Codex ran out of room in the model's context window.", 2, true],
|
||||
['usage', 'Usage limit exceeded', 1, false]
|
||||
] as const)('preserves %s failure handling during Reserve', async (_kind, error, attempts, compact) => {
|
||||
harness.resumeModel = 'gpt-reserve';
|
||||
harness.remainingThreadSystemErrors = 1;
|
||||
harness.nextThreadSystemErrorMessage = error;
|
||||
harness.reserveUsage = {
|
||||
accountId: 'account-a', ordinaryUsageAllowed: false,
|
||||
rateLimits: { limitId: 'codex', primary: { usedPercent: 100 } }
|
||||
};
|
||||
const { session } = createSessionStub(['first'], createMode(), true);
|
||||
session.sessionId = 'thread-reserve';
|
||||
await codexRemoteLauncher(session as never);
|
||||
expect(harness.startTurnMessages).toEqual(Array(attempts).fill('first'));
|
||||
expect(harness.compactThreadIds).toEqual(compact ? ['thread-reserve'] : []);
|
||||
expect(harness.startThreadIds).toEqual([]);
|
||||
expect(harness.startTurnParams[0]).toMatchObject({
|
||||
collaborationMode: { settings: { model: 'gpt-reserve' } }
|
||||
});
|
||||
});
|
||||
|
||||
it('reconciles a resumed Reserve task before waiting for the first user message', async () => {
|
||||
harness.resumeModel = 'gpt-reserve';
|
||||
harness.reserveUsage = {
|
||||
accountId: 'account-a', ordinaryUsageAllowed: false,
|
||||
rateLimits: { limitId: 'codex', primary: { usedPercent: 100 } },
|
||||
rateLimitsByLimitId: { base_model_inference: { limitName: 'gpt-reserve', primary: { usedPercent: 20 } } }
|
||||
};
|
||||
const { session, getModel, getAgentState } = createSessionStub([], createMode(), false, false);
|
||||
session.sessionId = 'thread-reserve';
|
||||
const run = codexRemoteLauncher(session as never);
|
||||
try {
|
||||
await vi.waitFor(() => expect(getAgentState().codexUsage?.reserve?.primary?.remainingPercent).toBe(80));
|
||||
expect(getModel()).toBe('gpt-5.6-luna');
|
||||
expect(harness.resumeThreadIds).toEqual(['thread-reserve']);
|
||||
expect(harness.startTurnMessages).toEqual([]);
|
||||
} finally {
|
||||
session.queue.close();
|
||||
await run;
|
||||
}
|
||||
});
|
||||
|
||||
it('reads native settings after transport loss without replaying the interrupted message', async () => {
|
||||
harness.resumeModel = 'gpt-reserve';
|
||||
harness.suppressTurnCompletion = true;
|
||||
harness.reserveUsage = {
|
||||
accountId: 'account-a', ordinaryUsageAllowed: false,
|
||||
rateLimits: { limitId: 'codex', primary: { usedPercent: 100 } }
|
||||
};
|
||||
const { session, getModel, getAgentState } = createSessionStub(['first'], createMode(), false, false);
|
||||
session.sessionId = 'thread-reserve';
|
||||
const run = codexRemoteLauncher(session as never);
|
||||
try {
|
||||
await vi.waitFor(() => expect(harness.startTurnMessages).toEqual(['first']));
|
||||
harness.resumeModel = 'gpt-5.4';
|
||||
harness.reserveUsage.ordinaryUsageAllowed = true;
|
||||
harness.appServerConnected = false;
|
||||
harness.abandonTransport?.();
|
||||
await vi.waitFor(() => expect(harness.resumeThreadIds).toEqual(['thread-reserve', 'thread-reserve']));
|
||||
await vi.waitFor(() => expect(getAgentState().codexUsage?.reserve).toBeNull());
|
||||
expect(getModel()).toBe('gpt-5.4');
|
||||
expect(harness.startTurnMessages).toEqual(['first']);
|
||||
} finally {
|
||||
session.queue.close();
|
||||
await run;
|
||||
}
|
||||
});
|
||||
|
||||
it('uses accepted Reserve settings for queued turns in the same conversation without replay', async () => {
|
||||
const home = await mkdtemp(join(tmpdir(), 'hapi-reserve-launcher-'));
|
||||
vi.stubEnv('CODEX_HOME', home);
|
||||
try {
|
||||
harness.reserveUsage = {
|
||||
accountId: 'account-a', ordinaryUsageAllowed: false,
|
||||
rateLimits: { limitId: 'codex', primary: { usedPercent: 100 } },
|
||||
rateLimitsByLimitId: { base_model_inference: { limitName: 'gpt-reserve', primary: { usedPercent: 20 } } },
|
||||
rateLimitUpsell: { banner_type: 'luna_reserve', blocked_model_slug: 'gpt-5.4' }
|
||||
};
|
||||
const { session, getModel, sessionEvents } = createSessionStub(['first', 'second'], createMode(), true);
|
||||
session.sessionId = '01900000-0000-7000-8000-000000000001';
|
||||
await codexRemoteLauncher(session as never);
|
||||
expect(harness.resumeThreadIds).toEqual([session.sessionId]);
|
||||
expect(harness.resumeThreadParams[0]).not.toHaveProperty('model');
|
||||
expect(harness.reserveSettingsCalls).toHaveLength(1);
|
||||
expect(harness.startTurnMessages).toEqual(['first', 'second']);
|
||||
expect(harness.startTurnParams).toHaveLength(2);
|
||||
for (const params of harness.startTurnParams) {
|
||||
expect(params.collaborationMode).toMatchObject({ settings: { model: 'gpt-reserve', reasoning_effort: 'medium' } });
|
||||
expect(params.threadId).toBe(session.sessionId);
|
||||
}
|
||||
expect(getModel()).toBe('gpt-5.6-luna');
|
||||
expect(sessionEvents.filter(event => String(event.message).includes('Luna Reserve'))).toHaveLength(1);
|
||||
} finally {
|
||||
vi.unstubAllEnvs();
|
||||
await rm(home, { recursive: true, force: true });
|
||||
}
|
||||
});
|
||||
|
||||
it('does not create a new thread when an existing conversation cannot be resumed', async () => {
|
||||
harness.failResumeThreadIds = ['thread-old'];
|
||||
const { session, sessionEvents } = createSessionStub(['first message']);
|
||||
|
||||
@@ -18,7 +18,6 @@ import { registerGeneratedImageFromPath } from '@/modules/common/generatedImages
|
||||
import { registerAppServerPermissionHandlers } from './utils/appServerPermissionAdapter';
|
||||
import {
|
||||
buildThreadStartParams,
|
||||
withoutCodexModelOverrides,
|
||||
buildTurnStartParams,
|
||||
type CodexContextManagementConfig
|
||||
} from './utils/appServerConfig';
|
||||
@@ -39,7 +38,6 @@ import {
|
||||
type RemoteLauncherExitReason
|
||||
} from '@/modules/common/remote/RemoteLauncherBase';
|
||||
import { CodexConversationHistory } from './conversationHistory';
|
||||
import { LunaReserve } from './utils/lunaReserve';
|
||||
|
||||
|
||||
|
||||
@@ -361,9 +359,6 @@ class CodexRemoteLauncher extends RemoteLauncherBase {
|
||||
});
|
||||
}
|
||||
|
||||
private lunaReserve: LunaReserve | null = null;
|
||||
private usageTimer: ReturnType<typeof setInterval> | null = null;
|
||||
|
||||
protected async runMainLoop(): Promise<void> {
|
||||
const session = this.session;
|
||||
const messageBuffer = this.messageBuffer;
|
||||
@@ -1867,7 +1862,6 @@ class CodexRemoteLauncher extends RemoteLauncherBase {
|
||||
}
|
||||
};
|
||||
|
||||
let hasThread = false;
|
||||
let activeMessage: QueuedMessage | null = null;
|
||||
let sameThreadRetryAttempt = 0;
|
||||
let sameThreadCompactAttempt = 0;
|
||||
@@ -1922,9 +1916,6 @@ class CodexRemoteLauncher extends RemoteLauncherBase {
|
||||
if (session.thinking) session.onThinkingChange(false);
|
||||
recoveryInFlight = false;
|
||||
activeMessage = null;
|
||||
hasThread = false;
|
||||
this.lunaReserve?.detach();
|
||||
this.currentThreadId = null;
|
||||
this.currentTurnId = null;
|
||||
wakeLoop();
|
||||
});
|
||||
@@ -2056,7 +2047,6 @@ class CodexRemoteLauncher extends RemoteLauncherBase {
|
||||
experimentalApi: true
|
||||
}
|
||||
});
|
||||
await this.lunaReserve?.initialize();
|
||||
};
|
||||
|
||||
// Returns 'accepted' when the thread contains the steered user message
|
||||
@@ -2920,9 +2910,7 @@ class CodexRemoteLauncher extends RemoteLauncherBase {
|
||||
const isThreadStatusFailure = msgType === 'task_failed' && msg.terminal_source === 'thread_status';
|
||||
const error = msgType === 'task_failed' ? asString(msg.error) : null;
|
||||
const explicitlyNonRetryable = msgType === 'task_failed'
|
||||
&& (msg.retryable === false || isPolicyBlockedCodexFailure(msg, error)
|
||||
|| msg.codex_error_info === 'usageLimitExceeded'
|
||||
|| /usage limit|rate limit|quota/i.test(error ?? ''));
|
||||
&& (msg.retryable === false || isPolicyBlockedCodexFailure(msg, error));
|
||||
|
||||
if (deferredThreadStatusFailure && isTerminalEvent && !isThreadStatusFailure) {
|
||||
const sameThread = !eventThreadId || eventThreadId === deferredThreadStatusFailure.threadId;
|
||||
@@ -3504,37 +3492,7 @@ class CodexRemoteLauncher extends RemoteLauncherBase {
|
||||
}
|
||||
};
|
||||
|
||||
const reserve = new LunaReserve(appServerClient, (model, effort, serviceTier) => {
|
||||
session.setModel(model);
|
||||
session.setModelReasoningEffort(effort);
|
||||
if (serviceTier !== undefined) session.setServiceTier(serviceTier);
|
||||
session.pushKeepAlive();
|
||||
}, (codexUsage) => {
|
||||
session.client.updateAgentState(state => ({ ...state, codexUsage }));
|
||||
}, message => session.sendSessionEvent({ type: 'message', message }));
|
||||
this.lunaReserve = reserve;
|
||||
const currentMode = (): EnhancedMode => ({
|
||||
permissionMode: 'default',
|
||||
model: session.getModel() ?? undefined,
|
||||
modelReasoningEffort: session.getModelReasoningEffort() ?? undefined,
|
||||
collaborationMode: session.getCollaborationMode() ?? 'default',
|
||||
serviceTier: session.getServiceTier()
|
||||
});
|
||||
const refreshUsage = () => reserve.refresh(currentMode(), () => !turnInFlight && !this.shouldExit);
|
||||
|
||||
appServerClient.setNotificationHandler((method, params) => {
|
||||
if (method === 'account/rateLimits/updated' || method === 'account/updated') {
|
||||
reserve.invalidate();
|
||||
void refreshUsage();
|
||||
}
|
||||
if (method === 'thread/settings/updated') {
|
||||
const record = asRecord(params);
|
||||
const threadId = asString(record?.threadId);
|
||||
if (threadId) reserve.onSettings(threadId, record?.threadSettings);
|
||||
}
|
||||
if (method === 'error' || method === 'turn/completed') {
|
||||
reserve.invalidate();
|
||||
}
|
||||
if (method === 'skills/changed') {
|
||||
void refreshNativeSkills(true).catch((error) => {
|
||||
logger.debug(`[Codex] failed to refresh skills: ${errorMessage(error)}`);
|
||||
@@ -3564,9 +3522,6 @@ class CodexRemoteLauncher extends RemoteLauncherBase {
|
||||
});
|
||||
}
|
||||
}
|
||||
if (method === 'error' || method === 'turn/completed') {
|
||||
void (codexEventQueue ?? Promise.resolve()).then(refreshUsage);
|
||||
}
|
||||
});
|
||||
|
||||
appServerClient.setStderrHandler((text) => {
|
||||
@@ -3624,10 +3579,6 @@ class CodexRemoteLauncher extends RemoteLauncherBase {
|
||||
}
|
||||
});
|
||||
|
||||
await reserve.initialize();
|
||||
this.usageTimer = setInterval(() => { void refreshUsage(); }, 60_000);
|
||||
this.usageTimer.unref();
|
||||
|
||||
let contextManagementConfig: CodexContextManagementConfig | undefined;
|
||||
try {
|
||||
const effectiveConfig = (await appServerClient.readConfig({
|
||||
@@ -3732,6 +3683,7 @@ class CodexRemoteLauncher extends RemoteLauncherBase {
|
||||
logger.debug(`[Codex] collaborationMode/list failed: ${errorMessage(error)}`);
|
||||
}
|
||||
|
||||
let hasThread = false;
|
||||
let pending: QueuedMessage | null = null;
|
||||
let suppressReadyForAdminCommand = false;
|
||||
|
||||
@@ -3825,7 +3777,7 @@ class CodexRemoteLauncher extends RemoteLauncherBase {
|
||||
try {
|
||||
const resumeResponse = await appServerClient.resumeThread({
|
||||
threadId: resumeCandidate,
|
||||
...withoutCodexModelOverrides(threadParams)
|
||||
...threadParams
|
||||
}, {
|
||||
signal: this.abortController.signal
|
||||
});
|
||||
@@ -3833,7 +3785,6 @@ class CodexRemoteLauncher extends RemoteLauncherBase {
|
||||
const resumeThread = resumeRecord ? asRecord(resumeRecord.thread) : null;
|
||||
const threadId = asString(resumeThread?.id) ?? resumeCandidate;
|
||||
applyResolvedModel(resumeRecord?.model);
|
||||
reserve.attach(threadId, resumeResponse);
|
||||
this.currentThreadId = threadId;
|
||||
this.conversationHistory.setThreadId(threadId);
|
||||
void this.conversationHistory.probeCapabilities().catch(() => {});
|
||||
@@ -3871,7 +3822,7 @@ class CodexRemoteLauncher extends RemoteLauncherBase {
|
||||
}
|
||||
};
|
||||
|
||||
const ensureThread = async (mode: EnhancedMode): Promise<string | null> => {
|
||||
const ensureThreadForGoal = async (mode: EnhancedMode): Promise<string | null> => {
|
||||
if (this.currentThreadId && this.currentThreadId !== invalidThreadId) {
|
||||
hasThread = true;
|
||||
return this.currentThreadId;
|
||||
@@ -3891,7 +3842,7 @@ class CodexRemoteLauncher extends RemoteLauncherBase {
|
||||
try {
|
||||
const resumeResponse = await appServerClient.resumeThread({
|
||||
threadId: resumeCandidate,
|
||||
...withoutCodexModelOverrides(threadParams)
|
||||
...threadParams
|
||||
}, {
|
||||
signal: this.abortController.signal
|
||||
});
|
||||
@@ -3899,7 +3850,6 @@ class CodexRemoteLauncher extends RemoteLauncherBase {
|
||||
const resumeThread = resumeRecord ? asRecord(resumeRecord.thread) : null;
|
||||
const threadId = asString(resumeThread?.id) ?? resumeCandidate;
|
||||
applyResolvedModel(resumeRecord?.model);
|
||||
reserve.attach(threadId, resumeResponse);
|
||||
this.currentThreadId = threadId;
|
||||
this.conversationHistory.setThreadId(threadId);
|
||||
void this.conversationHistory.probeCapabilities().catch(() => {});
|
||||
@@ -3907,8 +3857,8 @@ class CodexRemoteLauncher extends RemoteLauncherBase {
|
||||
hasThread = true;
|
||||
return threadId;
|
||||
} catch (error) {
|
||||
logger.warn(`[Codex] Failed to resume app-server thread ${resumeCandidate}`, error);
|
||||
sendVisibleStatus(`Codex conversation ${resumeCandidate} could not be resumed`);
|
||||
logger.warn(`[Codex] Failed to resume app-server thread ${resumeCandidate} for /goal`, error);
|
||||
sendVisibleStatus(`Goal failed: Codex conversation ${resumeCandidate} could not be resumed`);
|
||||
return null;
|
||||
}
|
||||
}
|
||||
@@ -3931,7 +3881,6 @@ class CodexRemoteLauncher extends RemoteLauncherBase {
|
||||
const thread = threadRecord ? asRecord(threadRecord.thread) : null;
|
||||
const threadId = asString(thread?.id);
|
||||
applyResolvedModel(threadRecord?.model);
|
||||
if (threadId) reserve.attach(threadId, threadResponse);
|
||||
if (!threadId) {
|
||||
throw new Error('app-server thread/start did not return thread.id');
|
||||
}
|
||||
@@ -3985,7 +3934,7 @@ class CodexRemoteLauncher extends RemoteLauncherBase {
|
||||
return true;
|
||||
}
|
||||
|
||||
const threadId = await ensureThread(message.mode);
|
||||
const threadId = await ensureThreadForGoal(message.mode);
|
||||
if (!threadId) {
|
||||
return true;
|
||||
}
|
||||
@@ -4062,7 +4011,6 @@ class CodexRemoteLauncher extends RemoteLauncherBase {
|
||||
invalidThreadId = null;
|
||||
hasThread = false;
|
||||
session.resetCodexThread();
|
||||
reserve.detach();
|
||||
sendVisibleStatus('Context was reset');
|
||||
return true;
|
||||
}
|
||||
@@ -4097,19 +4045,10 @@ class CodexRemoteLauncher extends RemoteLauncherBase {
|
||||
return true;
|
||||
};
|
||||
|
||||
const reconcileIdleThread = async () => {
|
||||
if (!hasThread && session.queue.size() === 0 && session.sessionId && session.sessionId !== session.sourceSessionId) {
|
||||
await ensureThread(currentMode());
|
||||
await refreshUsage();
|
||||
}
|
||||
};
|
||||
await reconcileIdleThread();
|
||||
|
||||
while (!this.shouldExit) {
|
||||
logActiveHandles('loop-top');
|
||||
if (!appServerClient.isConnected() || !appServerClient.isInitialized()) {
|
||||
await ensureAppServerInitialized();
|
||||
await reconcileIdleThread();
|
||||
}
|
||||
if (pendingSteerReconciliations.size > 0) {
|
||||
await runSteerReconciliation();
|
||||
@@ -4197,13 +4136,13 @@ class CodexRemoteLauncher extends RemoteLauncherBase {
|
||||
const response = shouldForkImportedSource
|
||||
? await appServerClient.forkThread({
|
||||
threadId: resumeCandidate,
|
||||
...withoutCodexModelOverrides(threadParams)
|
||||
...threadParams
|
||||
}, {
|
||||
signal: this.abortController.signal
|
||||
})
|
||||
: await appServerClient.resumeThread({
|
||||
threadId: resumeCandidate,
|
||||
...withoutCodexModelOverrides(threadParams)
|
||||
...threadParams
|
||||
}, {
|
||||
signal: this.abortController.signal
|
||||
});
|
||||
@@ -4211,7 +4150,6 @@ class CodexRemoteLauncher extends RemoteLauncherBase {
|
||||
const responseThread = responseRecord ? asRecord(responseRecord.thread) : null;
|
||||
threadId = asString(responseThread?.id) ?? resumeCandidate;
|
||||
applyResolvedModel(responseRecord?.model);
|
||||
if (threadId) reserve.attach(threadId, response, shouldForkImportedSource ? resumeCandidate : undefined);
|
||||
logger.debug(shouldForkImportedSource
|
||||
? `[Codex] Forked imported app-server thread ${resumeCandidate} -> ${threadId}`
|
||||
: `[Codex] Resumed app-server thread ${threadId}`);
|
||||
@@ -4237,7 +4175,6 @@ class CodexRemoteLauncher extends RemoteLauncherBase {
|
||||
const thread = threadRecord ? asRecord(threadRecord.thread) : null;
|
||||
threadId = asString(thread?.id);
|
||||
applyResolvedModel(threadRecord?.model);
|
||||
if (threadId) reserve.attach(threadId, threadResponse);
|
||||
if (!threadId) {
|
||||
throw new Error('app-server thread/start did not return thread.id');
|
||||
}
|
||||
@@ -4261,22 +4198,13 @@ class CodexRemoteLauncher extends RemoteLauncherBase {
|
||||
}
|
||||
}
|
||||
|
||||
const beforeRefresh = currentMode();
|
||||
await reserve.refresh(message.mode, () => !turnInFlight && !this.shouldExit);
|
||||
const afterRefresh = currentMode();
|
||||
const transitioned = beforeRefresh.model !== afterRefresh.model
|
||||
|| beforeRefresh.modelReasoningEffort !== afterRefresh.modelReasoningEffort
|
||||
|| beforeRefresh.serviceTier !== afterRefresh.serviceTier;
|
||||
setTurnInFlight(true);
|
||||
this.conversationHistory.setBusy(true);
|
||||
allowAnonymousTerminalEvent = false;
|
||||
const mode = reserve.turnMode({
|
||||
const mode = {
|
||||
...message.mode,
|
||||
model: transitioned ? afterRefresh.model : message.mode.model ?? afterRefresh.model,
|
||||
modelReasoningEffort: transitioned ? afterRefresh.modelReasoningEffort
|
||||
: message.mode.modelReasoningEffort ?? afterRefresh.modelReasoningEffort,
|
||||
serviceTier: transitioned ? afterRefresh.serviceTier : message.mode.serviceTier ?? afterRefresh.serviceTier
|
||||
});
|
||||
model: session.getModel() ?? message.mode.model
|
||||
};
|
||||
usageModel = typeof mode.model === 'string' && mode.model.trim()
|
||||
? mode.model.trim()
|
||||
: null;
|
||||
@@ -4414,10 +4342,6 @@ class CodexRemoteLauncher extends RemoteLauncherBase {
|
||||
|
||||
protected async cleanup(): Promise<void> {
|
||||
logger.debug('[codex-remote]: cleanup start');
|
||||
if (this.usageTimer) clearInterval(this.usageTimer);
|
||||
this.usageTimer = null;
|
||||
this.lunaReserve?.dispose();
|
||||
this.lunaReserve = null;
|
||||
this.appServerClient.setTransportAbandonedHandler(null);
|
||||
this.appServerClient.setStderrHandler(null);
|
||||
try {
|
||||
|
||||
@@ -31,7 +31,10 @@ vi.mock('../utils/buildHapiMcpBridge', () => ({ buildHapiMcpBridge: async () =>
|
||||
}) }));
|
||||
|
||||
const cleanups: Array<() => Promise<void>> = [];
|
||||
afterEach(async () => { for (const cleanup of cleanups.splice(0)) await cleanup(); });
|
||||
afterEach(async () => {
|
||||
try { for (const cleanup of cleanups.splice(0)) await cleanup(); }
|
||||
finally { vi.useRealTimers(); }
|
||||
});
|
||||
|
||||
async function fixture() {
|
||||
const directory = await mkdtemp('/tmp/hapi-shared-root-');
|
||||
@@ -45,7 +48,7 @@ async function fixture() {
|
||||
updateAgentState: updateState, keepAlive() {},
|
||||
onUserMessage() {}, onCancelQueuedMessage() {}, onRetryQueuedMessage() {},
|
||||
onReconnect: (fn: (() => void) | null) => { reconnect = fn; },
|
||||
rpcHandlerManager: { registerHandler() {} }, sendSessionEvent() {}, sendAgentMessage() {},
|
||||
rpcHandlerManager: { registerHandler() {} }, sendSessionEvent() {}, sendAgentMessage() {}, emitSessionReady() {},
|
||||
sendSessionDeath() {}, async flush() {}, close() {}
|
||||
} as unknown as ApiSessionClient;
|
||||
const root = new SharedCodexRoot({ session, workingDirectory: directory } as SessionBootstrapResult, {
|
||||
@@ -66,6 +69,21 @@ async function fixture() {
|
||||
}
|
||||
|
||||
describe('shared steering availability', () => {
|
||||
it('keeps idle sessions online without polling usage or publishing agent-state updates', async () => {
|
||||
const f = await fixture();
|
||||
const requests = vi.spyOn(f.root.client, 'request');
|
||||
const heartbeat = vi.spyOn(f.root.session, 'keepAlive');
|
||||
const updates = f.updateState.mock.calls.length;
|
||||
vi.useFakeTimers();
|
||||
|
||||
await f.root.activate();
|
||||
await vi.advanceTimersByTimeAsync(5 * 60_000);
|
||||
|
||||
expect(heartbeat).toHaveBeenCalled();
|
||||
expect(requests).not.toHaveBeenCalled();
|
||||
expect(f.updateState).toHaveBeenCalledTimes(updates);
|
||||
});
|
||||
|
||||
it('publishes root turn transitions, ignores child turns, and clears on shutdown', async () => {
|
||||
const f = await fixture();
|
||||
expect(f.state().steeringActive).toBe(false);
|
||||
|
||||
@@ -12,7 +12,6 @@ import { CodexAppServerClient } from '../codexAppServerClient';
|
||||
import { buildHapiMcpBridge, type HapiMcpBridge } from '../utils/buildHapiMcpBridge';
|
||||
import { buildUserInputFromMessage } from '../utils/appServerConfig';
|
||||
import { resolveCodexPermissionModeConfig } from '../utils/permissionModeConfig';
|
||||
import { LunaReserve } from '../utils/lunaReserve';
|
||||
import { resolveCodexSlashCommand } from '../utils/slashCommands';
|
||||
import { parseReasoningEffortValue } from '../utils/reasoningEffort';
|
||||
import { SharedCodexPermissions } from './permissions';
|
||||
@@ -49,9 +48,7 @@ export class SharedCodexRoot {
|
||||
private projection!: SharedCodexProjection;
|
||||
private readonly children = new Map<string, SharedCodexProjection>();
|
||||
private readonly ancestry = new Map<string, string | null>();
|
||||
private readonly reserve: LunaReserve;
|
||||
private heartbeat?: ReturnType<typeof setInterval>;
|
||||
private usageTimer?: ReturnType<typeof setInterval>;
|
||||
private work: Promise<unknown> = Promise.resolve();
|
||||
private notifications = Promise.resolve();
|
||||
private currentTurn: string | undefined;
|
||||
@@ -75,9 +72,6 @@ export class SharedCodexRoot {
|
||||
constructor(readonly bootstrap: SessionBootstrapResult, private readonly host: RootHost) {
|
||||
this.session = bootstrap.session;
|
||||
this.client = new CodexAppServerClient({ endpoint: host.endpoint, token: host.token, cwd: bootstrap.workingDirectory });
|
||||
this.reserve = new LunaReserve(this.client,
|
||||
(model, effort, serviceTier) => { this.settings = { ...this.settings, model, modelReasoningEffort: effort, serviceTier }; },
|
||||
codexUsage => this.session.updateAgentState(state => ({ ...state, codexUsage })), message => this.notice(message));
|
||||
this.client.setNotificationHandler((method, params) => {
|
||||
if (method === 'serverRequest/resolved') {
|
||||
this.permissions?.resolved(string(record(params).threadId) ?? '', record(params).requestId); return;
|
||||
@@ -105,7 +99,6 @@ export class SharedCodexRoot {
|
||||
const text = formatMessageWithAttachments(message.content.text, message.content.attachments);
|
||||
const resolved = text.trim().startsWith('/') ? await this.queue.command(id, () => this.command(text)) : text;
|
||||
if (resolved === null) { this.session.emitMessagesConsumed([id], { clearQueuedThinkingGrace: true }); return; }
|
||||
await this.refreshUsage();
|
||||
await this.queue.enqueue(id, buildUserInputFromMessage(resolved), this.interrupted);
|
||||
}).catch(error => this.notice(`Message not confirmed: ${error instanceof Error ? error.message : error}. Inspect the queue before retrying.`));
|
||||
});
|
||||
@@ -186,7 +179,7 @@ export class SharedCodexRoot {
|
||||
[id, { ...request, completedAt: Date.now(), status: 'canceled' as const }])) }
|
||||
}));
|
||||
if (subscribe) response = record(await this.client.request('thread/resume', { threadId }));
|
||||
this.acceptSettings(response); this.acceptSettings(this.host.settingsFor(threadId) ?? {}); this.reserve.attach(threadId, response);
|
||||
this.acceptSettings(response); this.acceptSettings(this.host.settingsFor(threadId) ?? {});
|
||||
await this.projection.history(response.thread); await this.refresh(); await this.refreshChildren(true);
|
||||
}
|
||||
async activate(options: SharedLaunchOptions = {}): Promise<void> {
|
||||
@@ -195,8 +188,6 @@ export class SharedCodexRoot {
|
||||
if (this.stopping) throw new Error('Codex execution is stopping');
|
||||
this.registerControls(); this.ready();
|
||||
this.heartbeat = setInterval(() => this.alive(), 2_000); this.alive(); this.session.emitSessionReady();
|
||||
void this.reserve.initialize().catch(error => logger.debug('[Codex shared] usage initialization', error));
|
||||
this.usageTimer = setInterval(() => { void this.refreshUsage(); }, 60_000);
|
||||
}
|
||||
private publishSteering(): void {
|
||||
const active = Boolean(this.currentTurn) && !this.stopping && !this.closed && !this.reconnecting && this.client.isInitialized();
|
||||
@@ -220,7 +211,6 @@ export class SharedCodexRoot {
|
||||
...('serviceTier' in value ? { serviceTier: value.serviceTier === 'priority' ? 'fast' : 'standard' } : {}),
|
||||
...('collaborationMode' in value ? { collaborationMode: record(value.collaborationMode).mode === 'plan' ? 'plan' as const : 'default' as const } : {})
|
||||
};
|
||||
this.reserve.onSettings(this.threadId, value);
|
||||
for (const listener of this.settingsListeners) listener();
|
||||
}
|
||||
private async notification(method: string, params: unknown, modelAtReceipt?: string): Promise<void> {
|
||||
@@ -303,11 +293,6 @@ export class SharedCodexRoot {
|
||||
} catch (error) { logger.debug('[Codex shared] child history unavailable', { id, error }); }
|
||||
}
|
||||
}
|
||||
private refreshUsage(): Promise<void> {
|
||||
return this.reserve.refresh({ permissionMode: this.settings.permissionMode === 'yolo' ? 'yolo' : 'default',
|
||||
collaborationMode: this.settings.collaborationMode ?? 'default', model: this.settings.model ?? undefined },
|
||||
() => !this.closed && !this.currentTurn);
|
||||
}
|
||||
nativeQueueDeleted(nativeId: string): Promise<void> { return this.queue.deleted(nativeId); }
|
||||
replaySettings(): Array<{ method: string; params: unknown }> {
|
||||
return this.settingsNotification ? [{ method: 'thread/settings/updated', params: {
|
||||
@@ -482,9 +467,9 @@ export class SharedCodexRoot {
|
||||
return slash.kind === 'replace' ? slash.text : null;
|
||||
}
|
||||
stopAccepting(): void {
|
||||
this.stopping = true; this.ready(); clearInterval(this.usageTimer);
|
||||
this.stopping = true; this.ready();
|
||||
this.publishSteering();
|
||||
this.session.onReconnect(null); this.client.setTransportAbandonedHandler(null); this.reserve.dispose();
|
||||
this.session.onReconnect(null); this.client.setTransportAbandonedHandler(null);
|
||||
}
|
||||
async suspend(): Promise<void> {
|
||||
this.stopAccepting();
|
||||
|
||||
@@ -332,13 +332,3 @@ export function buildTurnStartParams(args: {
|
||||
|
||||
return params;
|
||||
}
|
||||
|
||||
/** Resume the actual task settings before reconciling HAPI's queued configuration. */
|
||||
export function withoutCodexModelOverrides(params: ThreadStartParams): ThreadStartParams {
|
||||
const { model, serviceTier, ...rest } = params;
|
||||
const config = { ...rest.config };
|
||||
delete config.model;
|
||||
delete config.model_reasoning_effort;
|
||||
delete config.service_tier;
|
||||
return { ...rest, config };
|
||||
}
|
||||
|
||||
@@ -1,175 +0,0 @@
|
||||
import { afterEach, describe, expect, it, vi } from 'vitest';
|
||||
import { mkdtemp, readFile, rm } from 'node:fs/promises';
|
||||
import { tmpdir } from 'node:os';
|
||||
import { join } from 'node:path';
|
||||
import { LunaReserve } from './lunaReserve';
|
||||
import { withoutCodexModelOverrides } from './appServerConfig';
|
||||
import type { EnhancedMode } from '../loop';
|
||||
|
||||
const threadId = '01900000-0000-7000-8000-000000000001';
|
||||
const mode: EnhancedMode = { permissionMode: 'yolo', collaborationMode: 'plan', model: 'gpt-5.6', modelReasoningEffort: 'high' };
|
||||
const dirs: string[] = [];
|
||||
afterEach(async () => { await Promise.all(dirs.splice(0).map(dir => rm(dir, { recursive: true, force: true }))); });
|
||||
|
||||
async function fixture() {
|
||||
const dir = await mkdtemp(join(tmpdir(), 'hapi-reserve-'));
|
||||
dirs.push(dir);
|
||||
const response = {
|
||||
accountId: 'account-a', ordinaryUsageAllowed: true as boolean | null,
|
||||
rateLimits: { limitId: 'codex', primary: { usedPercent: 40 }, secondary: { usedPercent: 60 }, spendControlReached: false, rateLimitReachedType: null as string | null },
|
||||
rateLimitsByLimitId: {
|
||||
base_model_inference: { limitId: 'base_model_inference', limitName: 'gpt-reserve', primary: { usedPercent: 10 as unknown, resetsAt: 2000000000, windowDurationMins: 123 }, secondary: { usedPercent: 25 } }
|
||||
},
|
||||
rateLimitUpsell: null as unknown
|
||||
};
|
||||
const client = {
|
||||
readAccountRateLimits: vi.fn(async () => structuredClone(response) as unknown),
|
||||
supportsMethod: vi.fn(async () => true),
|
||||
listModels: vi.fn(async () => ({ data: [
|
||||
{ id: 'gpt-reserve', hidden: true, defaultReasoningEffort: 'medium', supportedReasoningEfforts: [{ reasoningEffort: 'medium' }] },
|
||||
{ id: 'gpt-5.6', hidden: false, defaultReasoningEffort: 'medium', supportedReasoningEfforts: [{ reasoningEffort: 'high' }] }
|
||||
] })),
|
||||
updateThreadSettings: vi.fn(async () => {})
|
||||
};
|
||||
const sync = vi.fn(); const publish = vi.fn(); const notice = vi.fn();
|
||||
const reserve = new LunaReserve(client, sync, publish, notice, dir);
|
||||
await reserve.initialize();
|
||||
reserve.attach(threadId, { model: 'gpt-5.6', reasoningEffort: 'high', serviceTier: 'priority' });
|
||||
const enter = async () => {
|
||||
response.ordinaryUsageAllowed = false;
|
||||
response.rateLimitUpsell = { banner_type: 'luna_reserve', blocked_model_slug: 'gpt-5.6' };
|
||||
await reserve.refresh(mode, () => true);
|
||||
};
|
||||
const usage = () => publish.mock.lastCall?.[0];
|
||||
return { dir, response, client, reserve, sync, publish, notice, enter, usage };
|
||||
}
|
||||
|
||||
describe('Luna Reserve authoritative transitions', () => {
|
||||
it('hides an available bucket and does not infer eligibility from exhausted percentages', async () => {
|
||||
const f = await fixture();
|
||||
f.response.rateLimits.primary.usedPercent = 100;
|
||||
f.response.rateLimits.secondary.usedPercent = 100;
|
||||
await f.reserve.refresh(mode, () => true);
|
||||
expect(f.usage().reserve).toBeNull();
|
||||
expect(f.client.updateThreadSettings).not.toHaveBeenCalled();
|
||||
f.response.ordinaryUsageAllowed = false;
|
||||
await f.reserve.refresh(mode, () => true);
|
||||
expect(f.client.updateThreadSettings).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it('activates only after acceptance, separates buckets, preserves return settings, and hides on recovery', async () => {
|
||||
const f = await fixture();
|
||||
await f.enter();
|
||||
expect(f.client.readAccountRateLimits).toHaveBeenLastCalledWith(true);
|
||||
expect(f.client.updateThreadSettings).toHaveBeenCalledWith({ threadId, model: 'gpt-reserve', effort: 'medium', serviceTier: null,
|
||||
collaborationMode: { mode: 'plan', settings: { model: 'gpt-reserve', reasoning_effort: 'medium', developer_instructions: null } } });
|
||||
expect(f.usage().ordinary.secondary.remainingPercent).toBe(40);
|
||||
expect(f.usage().reserve.secondary.remainingPercent).toBe(75);
|
||||
expect(f.usage().reserve.primary).toEqual({ remainingPercent: 90, resetsAt: 2000000000, windowDurationMins: 123 });
|
||||
expect(f.sync).toHaveBeenLastCalledWith('gpt-5.6-luna', 'medium', 'standard');
|
||||
expect(JSON.parse(await readFile(join(f.dir, 'tui-luna-reserve', `${threadId}.json`), 'utf8'))).toEqual({ account_id: 'account-a', model: 'gpt-5.6', effort: 'high' });
|
||||
expect(f.reserve.turnMode(mode)).toMatchObject({ model: 'gpt-reserve', modelReasoningEffort: 'medium', serviceTier: 'standard' });
|
||||
f.response.ordinaryUsageAllowed = true;
|
||||
f.response.rateLimitUpsell = null;
|
||||
await f.reserve.refresh(mode, () => true);
|
||||
expect(f.sync).toHaveBeenLastCalledWith('gpt-5.6', 'high', 'standard');
|
||||
expect(f.usage().reserve).toBeNull();
|
||||
expect(f.notice).toHaveBeenCalledTimes(2);
|
||||
});
|
||||
|
||||
it('retains exhausted Reserve without another switch or retry and treats missing/invalid usage as unknown', async () => {
|
||||
const f = await fixture(); await f.enter();
|
||||
for (const used of [undefined, null, '50', NaN, Infinity, -1, 101, 100]) {
|
||||
f.response.rateLimitsByLimitId.base_model_inference.primary.usedPercent = used;
|
||||
await f.reserve.refresh(mode, () => true);
|
||||
expect(f.usage().reserve.primary.remainingPercent).toBe(used === 100 ? 0 : null);
|
||||
}
|
||||
expect(f.client.updateThreadSettings).toHaveBeenCalledTimes(1);
|
||||
});
|
||||
|
||||
it('requires a fresh account-bound recovery read; sparse notifications and unknown banners cannot recover', async () => {
|
||||
const f = await fixture(); await f.enter();
|
||||
f.response.ordinaryUsageAllowed = true;
|
||||
f.response.rateLimitUpsell = { banner_type: 'unknown_future_banner' };
|
||||
await f.reserve.refresh(mode, () => true);
|
||||
f.response.rateLimitUpsell = null;
|
||||
f.response.accountId = 'account-b';
|
||||
await f.reserve.refresh(mode, () => true);
|
||||
expect(f.client.updateThreadSettings).toHaveBeenCalledTimes(1);
|
||||
f.response.accountId = 'account-a';
|
||||
let finish!: (value: unknown) => void;
|
||||
f.client.readAccountRateLimits.mockImplementationOnce(() => new Promise(resolve => { finish = resolve; }));
|
||||
const pending = f.reserve.refresh(mode, () => true);
|
||||
await vi.waitFor(() => expect(finish).toBeTypeOf('function'));
|
||||
f.reserve.invalidate();
|
||||
finish(f.response); await pending;
|
||||
expect(f.client.updateThreadSettings).toHaveBeenCalledTimes(1);
|
||||
});
|
||||
|
||||
it('reconciles a resumed Reserve task with the TUI return target and ignores another thread settings', async () => {
|
||||
const f = await fixture(); await f.enter(); f.reserve.dispose();
|
||||
const resumed = new LunaReserve(f.client, f.sync, f.publish, f.notice, f.dir);
|
||||
await resumed.initialize();
|
||||
resumed.attach(threadId, { model: 'gpt-reserve', reasoningEffort: 'medium' });
|
||||
resumed.onSettings('other-thread', { model: 'gpt-5.6', effort: 'high' });
|
||||
expect(resumed.turnMode(mode).model).toBe('gpt-reserve');
|
||||
f.response.ordinaryUsageAllowed = true; f.response.rateLimitUpsell = null;
|
||||
await resumed.refresh(mode, () => true);
|
||||
expect(f.sync).toHaveBeenLastCalledWith('gpt-5.6', 'high', 'standard');
|
||||
expect(f.usage().reserve).toBeNull();
|
||||
});
|
||||
|
||||
it('copies the native fork return target without deleting the parent recovery record', async () => {
|
||||
const f = await fixture(); await f.enter();
|
||||
const forkId = '01900000-0000-7000-8000-000000000002';
|
||||
f.reserve.attach(forkId, { model: 'gpt-reserve', reasoningEffort: 'medium', thread: { forkedFromId: threadId } });
|
||||
await f.reserve.refresh(mode, () => true);
|
||||
const parent = join(f.dir, 'tui-luna-reserve', `${threadId}.json`);
|
||||
const child = join(f.dir, 'tui-luna-reserve', `${forkId}.json`);
|
||||
expect(await readFile(child, 'utf8')).toBe(await readFile(parent, 'utf8'));
|
||||
f.response.ordinaryUsageAllowed = true; f.response.rateLimitUpsell = null;
|
||||
await f.reserve.refresh(mode, () => true);
|
||||
expect(f.client.updateThreadSettings).toHaveBeenLastCalledWith(expect.objectContaining({ threadId: forkId, model: 'gpt-5.6' }));
|
||||
expect(JSON.parse(await readFile(parent, 'utf8')).model).toBe('gpt-5.6');
|
||||
await expect(readFile(child, 'utf8')).rejects.toThrow();
|
||||
f.reserve.detach();
|
||||
expect(f.usage()).toBeNull();
|
||||
});
|
||||
|
||||
it('keeps selection and visibility unchanged on a rejected settings update without looping', async () => {
|
||||
const f = await fixture();
|
||||
f.client.updateThreadSettings.mockRejectedValue(new Error('Method not found'));
|
||||
await f.enter(); await f.reserve.refresh(mode, () => true);
|
||||
expect(f.client.updateThreadSettings).toHaveBeenCalledTimes(1);
|
||||
expect(f.usage().reserve).toBeNull();
|
||||
expect(f.reserve.turnMode(mode).model).toBe('gpt-5.6');
|
||||
expect(f.notice).toHaveBeenCalledTimes(1);
|
||||
});
|
||||
|
||||
it('does not advertise or activate Reserve on an older protocol', async () => {
|
||||
const f = await fixture();
|
||||
f.client.readAccountRateLimits.mockImplementation(async () => ({ rateLimits: f.response.rateLimits, accountId: 'account-a', rateLimitUpsell: { banner_type: 'luna_reserve' } }));
|
||||
const old = new LunaReserve(f.client, f.sync, f.publish, f.notice, f.dir);
|
||||
await old.initialize(); old.attach(threadId, { model: 'gpt-5.6' });
|
||||
await old.refresh(mode, () => true);
|
||||
expect(f.client.readAccountRateLimits).toHaveBeenLastCalledWith(false);
|
||||
expect(f.client.updateThreadSettings).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it('waits until idle and hides state after exit/disposal', async () => {
|
||||
const f = await fixture();
|
||||
f.response.ordinaryUsageAllowed = false; f.response.rateLimitUpsell = { banner_type: 'luna_reserve' };
|
||||
await f.reserve.refresh(mode, () => false);
|
||||
expect(f.client.updateThreadSettings).not.toHaveBeenCalled();
|
||||
await f.enter();
|
||||
f.reserve.onSettings(threadId, { model: 'gpt-5.6', effort: 'high' });
|
||||
expect(f.usage().reserve).toBeNull();
|
||||
f.reserve.dispose();
|
||||
expect(f.usage()).toBeNull();
|
||||
});
|
||||
|
||||
it('omits model overrides on resume while retaining permissions, tools and instructions', () => {
|
||||
expect(withoutCodexModelOverrides({ model: 'stale', serviceTier: 'priority', sandbox: 'workspace-write', config: { model_reasoning_effort: 'high', 'mcp_servers.hapi': {}, developer_instructions: 'hapi' } }))
|
||||
.toEqual({ sandbox: 'workspace-write', config: { 'mcp_servers.hapi': {}, developer_instructions: 'hapi' } });
|
||||
});
|
||||
});
|
||||
@@ -1,301 +0,0 @@
|
||||
import { mkdir, open, rename, rm } from 'node:fs/promises';
|
||||
import { homedir } from 'node:os';
|
||||
import { join } from 'node:path';
|
||||
import { randomUUID } from 'node:crypto';
|
||||
import { z } from 'zod';
|
||||
import { logger } from '@/ui/logger';
|
||||
import type { AgentState } from '@hapi/protocol';
|
||||
import type { CodexAppServerClient } from '../codexAppServerClient';
|
||||
import type { EnhancedMode } from '../loop';
|
||||
import type { ModelListItem, ThreadSettingsUpdateParams } from '../appServerTypes';
|
||||
|
||||
// This quota alias belongs only at the Codex boundary, never in HAPI's model picker.
|
||||
const RESERVE_MODEL = 'gpt-reserve';
|
||||
const LUNA_MODEL = 'gpt-5.6-luna';
|
||||
const WindowSchema = z.object({
|
||||
usedPercent: z.unknown().optional(),
|
||||
windowDurationMins: z.unknown().optional(),
|
||||
resetsAt: z.unknown().optional()
|
||||
});
|
||||
const BucketSchema = z.object({
|
||||
limitId: z.string().nullish(),
|
||||
limitName: z.string().nullish(),
|
||||
normalModelSlug: z.string().nullish(),
|
||||
primary: WindowSchema.nullish(),
|
||||
secondary: WindowSchema.nullish(),
|
||||
credits: z.object({ hasCredits: z.boolean(), unlimited: z.boolean() }).nullish(),
|
||||
spendControlReached: z.boolean().nullish(),
|
||||
rateLimitReachedType: z.string().nullish()
|
||||
});
|
||||
const UsageSchema = z.object({
|
||||
accountId: z.string().nullish(),
|
||||
ordinaryUsageAllowed: z.boolean().nullish(),
|
||||
rateLimits: BucketSchema,
|
||||
rateLimitsByLimitId: z.record(z.string(), BucketSchema).nullish(),
|
||||
rateLimitUpsell: z.unknown().optional()
|
||||
});
|
||||
const BannerSchema = z.object({
|
||||
banner_type: z.literal('luna_reserve'),
|
||||
blocked_model_slug: z.string().min(1).max(256).nullish()
|
||||
});
|
||||
const ReturnSchema = z.object({
|
||||
account_id: z.string().min(1),
|
||||
model: z.string().min(1).max(256),
|
||||
effort: z.string().nullable()
|
||||
});
|
||||
const SettingsSchema = z.object({
|
||||
model: z.string().min(1),
|
||||
effort: z.string().nullish(),
|
||||
serviceTier: z.string().nullish()
|
||||
});
|
||||
type Usage = z.infer<typeof UsageSchema>;
|
||||
type Bucket = z.infer<typeof BucketSchema>;
|
||||
type Settings = z.infer<typeof SettingsSchema>;
|
||||
type ReserveClient = Pick<CodexAppServerClient, 'readAccountRateLimits' | 'listModels' | 'supportsMethod' | 'updateThreadSettings'>;
|
||||
|
||||
function usageBucket(bucket?: Bucket | null): NonNullable<AgentState['codexUsage']>['ordinary'] {
|
||||
const window = (value: Bucket['primary']) => {
|
||||
if (!value) return null;
|
||||
const finite = (n: unknown): n is number => typeof n === 'number' && Number.isFinite(n);
|
||||
return {
|
||||
remainingPercent: finite(value.usedPercent) && value.usedPercent >= 0 && value.usedPercent <= 100
|
||||
? 100 - value.usedPercent : null,
|
||||
windowDurationMins: finite(value.windowDurationMins) && value.windowDurationMins > 0 ? value.windowDurationMins : null,
|
||||
resetsAt: finite(value.resetsAt) && value.resetsAt >= 0 ? value.resetsAt : null
|
||||
};
|
||||
};
|
||||
return { primary: window(bucket?.primary), secondary: window(bucket?.secondary) };
|
||||
}
|
||||
|
||||
/** Account reads authorize transitions; settings accepted by the thread establish activation. */
|
||||
export class LunaReserve {
|
||||
private threadId: string | null = null;
|
||||
private settings: Settings | null = null;
|
||||
private settingsRevision = 0;
|
||||
private forkedFrom: string | null = null;
|
||||
private snapshot: Usage | null = null;
|
||||
private models: ModelListItem[] = [];
|
||||
private supported = false;
|
||||
private revision = 0;
|
||||
private disposed = false;
|
||||
private queue: Promise<void> = Promise.resolve();
|
||||
|
||||
constructor(
|
||||
private readonly client: ReserveClient,
|
||||
private readonly sync: (model: string, effort: string | null, serviceTier: string | null | undefined) => void,
|
||||
private readonly publish: (usage: AgentState['codexUsage']) => void,
|
||||
private readonly notice: (message: string) => void,
|
||||
private readonly codexHome = process.env.CODEX_HOME || join(homedir(), '.codex')
|
||||
) {}
|
||||
|
||||
async initialize(): Promise<void> {
|
||||
this.supported = false;
|
||||
this.models = [];
|
||||
try {
|
||||
const response = UsageSchema.parse(await this.client.readAccountRateLimits());
|
||||
this.snapshot = response;
|
||||
// Field presence is a protocol capability, not account eligibility.
|
||||
if ('ordinaryUsageAllowed' in response && await this.client.supportsMethod('thread/settings/update')) {
|
||||
let cursor: string | undefined;
|
||||
do {
|
||||
const page = await this.client.listModels({ includeHidden: true, ...(cursor ? { cursor } : {}) });
|
||||
this.models.push(...(page.data ?? []));
|
||||
cursor = page.nextCursor ?? undefined;
|
||||
} while (cursor);
|
||||
this.supported = this.models.some(model => (model.model ?? model.id) === RESERVE_MODEL);
|
||||
}
|
||||
this.publishUsage();
|
||||
} catch {
|
||||
// Older app-servers and API-key accounts can lack the usage endpoint.
|
||||
this.publish(null);
|
||||
}
|
||||
}
|
||||
|
||||
attach(threadId: string, response: unknown, forkedFrom?: string): void {
|
||||
const parsed = z.object({ model: z.string(), reasoningEffort: z.string().nullish(), serviceTier: z.string().nullish(), thread: z.object({ forkedFromId: z.string().nullish() }).optional() }).safeParse(response);
|
||||
this.threadId = threadId;
|
||||
this.forkedFrom = forkedFrom ?? (parsed.success ? parsed.data.thread?.forkedFromId : null) ?? null;
|
||||
this.invalidate();
|
||||
this.snapshot = null;
|
||||
if (parsed.success) {
|
||||
this.acceptSettings({ ...parsed.data, effort: parsed.data.reasoningEffort });
|
||||
} else {
|
||||
this.settings = null;
|
||||
}
|
||||
this.publishUsage();
|
||||
}
|
||||
|
||||
onSettings(threadId: string, value: unknown): void {
|
||||
if (threadId !== this.threadId || this.disposed) return;
|
||||
const parsed = SettingsSchema.safeParse(value);
|
||||
if (!parsed.success) return;
|
||||
this.invalidate();
|
||||
this.settingsRevision++;
|
||||
this.acceptSettings(parsed.data);
|
||||
this.publishUsage();
|
||||
}
|
||||
|
||||
private acceptSettings(settings: Settings): void {
|
||||
this.settings = settings;
|
||||
this.sync(settings.model === RESERVE_MODEL ? LUNA_MODEL : settings.model, settings.effort ?? null,
|
||||
settings.serviceTier === 'priority' ? 'fast' : settings.serviceTier === null ? 'standard' : undefined);
|
||||
}
|
||||
|
||||
invalidate(): void {
|
||||
this.revision++;
|
||||
}
|
||||
|
||||
detach(): void {
|
||||
this.invalidate();
|
||||
this.threadId = null;
|
||||
this.settings = null;
|
||||
this.snapshot = null;
|
||||
this.publishUsage();
|
||||
}
|
||||
|
||||
dispose(): void {
|
||||
this.disposed = true;
|
||||
this.invalidate();
|
||||
this.publish(null);
|
||||
}
|
||||
|
||||
private publishUsage(): void {
|
||||
const response = this.snapshot;
|
||||
if (!response || this.disposed) {
|
||||
this.publish(null);
|
||||
return;
|
||||
}
|
||||
// Never substitute a model-specific bucket for ordinary account windows.
|
||||
const ordinary = response.rateLimitsByLimitId?.codex
|
||||
?? (!response.rateLimits.limitId || response.rateLimits.limitId === 'codex' ? response.rateLimits : null);
|
||||
const reserve = Object.values(response.rateLimitsByLimitId ?? {}).find(bucket => bucket.limitName === RESERVE_MODEL)
|
||||
?? (response.rateLimits.limitName === RESERVE_MODEL ? response.rateLimits : null);
|
||||
this.publish({
|
||||
ordinary: usageBucket(ordinary),
|
||||
reserve: this.settings?.model === RESERVE_MODEL && response.accountId
|
||||
? usageBucket(reserve) : null
|
||||
});
|
||||
}
|
||||
|
||||
/** Serialize polling with pre-submission reconciliation; never submit or replay a message here. */
|
||||
refresh(mode: EnhancedMode, canSwitch: () => boolean): Promise<void> {
|
||||
this.queue = this.queue.then(() => this.readAndReconcile(mode, canSwitch));
|
||||
return this.queue;
|
||||
}
|
||||
|
||||
private returnPath(threadId = this.threadId): string | null {
|
||||
if (!z.string().uuid().safeParse(threadId).success) return null;
|
||||
// Shared with the official TUI so local/remote handoff preserves the task's return target.
|
||||
return join(this.codexHome, 'tui-luna-reserve', `${threadId}.json`);
|
||||
}
|
||||
|
||||
private async loadReturn(threadId = this.threadId): Promise<z.infer<typeof ReturnSchema> | null> {
|
||||
const path = this.returnPath(threadId);
|
||||
if (!path) return null;
|
||||
try {
|
||||
const handle = await open(path, 'r');
|
||||
try {
|
||||
if ((await handle.stat()).size > 4096) return null;
|
||||
return ReturnSchema.parse(JSON.parse(await handle.readFile('utf8')));
|
||||
} finally {
|
||||
await handle.close();
|
||||
}
|
||||
} catch { return null; }
|
||||
}
|
||||
|
||||
private async saveReturn(value: z.infer<typeof ReturnSchema>): Promise<void> {
|
||||
const path = this.returnPath();
|
||||
if (!path) throw new Error('Invalid Codex thread ID');
|
||||
await mkdir(join(this.codexHome, 'tui-luna-reserve'), { recursive: true });
|
||||
const temporary = `${path}.${randomUUID()}.tmp`;
|
||||
try {
|
||||
const handle = await open(temporary, 'wx', 0o600);
|
||||
try {
|
||||
await handle.writeFile(JSON.stringify(value));
|
||||
await handle.sync();
|
||||
} finally { await handle.close(); }
|
||||
await rename(temporary, path);
|
||||
} finally { await rm(temporary, { force: true }); }
|
||||
}
|
||||
|
||||
private async readAndReconcile(mode: EnhancedMode, canSwitch: () => boolean): Promise<void> {
|
||||
if (this.disposed) return;
|
||||
const revision = this.revision;
|
||||
let applying = false;
|
||||
try {
|
||||
const response = UsageSchema.parse(await this.client.readAccountRateLimits(this.supported));
|
||||
if (this.disposed || revision !== this.revision) return;
|
||||
this.snapshot = response;
|
||||
this.publishUsage();
|
||||
if (!this.supported || !this.threadId || !this.settings || !response.accountId || !canSwitch()) return;
|
||||
const current = this.settings;
|
||||
const inReserve = current.model === RESERVE_MODEL;
|
||||
let previous = inReserve ? await this.loadReturn() : null;
|
||||
if (inReserve && !previous && this.forkedFrom) {
|
||||
const inherited = await this.loadReturn(this.forkedFrom);
|
||||
if (inherited?.account_id === response.accountId) {
|
||||
await this.saveReturn(inherited);
|
||||
previous = inherited;
|
||||
}
|
||||
}
|
||||
const ordinary = response.rateLimits;
|
||||
const recovered = response.ordinaryUsageAllowed != null
|
||||
&& (response.ordinaryUsageAllowed || ordinary.credits?.hasCredits || ordinary.credits?.unlimited)
|
||||
&& response.rateLimitUpsell == null
|
||||
&& ordinary.spendControlReached !== true && ordinary.rateLimitReachedType == null;
|
||||
const banner = BannerSchema.safeParse(response.rateLimitUpsell);
|
||||
const entering = !inReserve && (!mode.model || mode.model === current.model)
|
||||
&& response.ordinaryUsageAllowed === false && banner.success
|
||||
&& (!banner.data.blocked_model_slug || banner.data.blocked_model_slug === current.model);
|
||||
const targetId = entering ? RESERVE_MODEL
|
||||
: inReserve && recovered && previous?.account_id === response.accountId ? previous.model : null;
|
||||
const target = this.models.find(model => (model.model ?? model.id) === targetId && (entering || model.hidden !== true));
|
||||
if (!target || !targetId || revision !== this.revision || !canSwitch()) return;
|
||||
const desiredEffort = entering ? current.effort : previous?.effort;
|
||||
const effort = target.supportedReasoningEfforts?.some(item => item.reasoningEffort === desiredEffort)
|
||||
? desiredEffort : target.defaultReasoningEffort;
|
||||
applying = true;
|
||||
if (entering) await this.saveReturn({ account_id: response.accountId, model: current.model, effort: current.effort ?? null });
|
||||
if (this.disposed || revision !== this.revision || !canSwitch()) return;
|
||||
const params: ThreadSettingsUpdateParams = {
|
||||
threadId: this.threadId,
|
||||
model: targetId,
|
||||
effort: effort ?? null,
|
||||
serviceTier: null,
|
||||
...(mode.collaborationMode ? { collaborationMode: {
|
||||
mode: mode.collaborationMode,
|
||||
settings: { model: targetId, reasoning_effort: effort ?? null, developer_instructions: null }
|
||||
} } : {})
|
||||
};
|
||||
const settingsRevision = this.settingsRevision;
|
||||
await this.client.updateThreadSettings(params);
|
||||
if (this.disposed || params.threadId !== this.threadId) return;
|
||||
if (this.settingsRevision !== settingsRevision
|
||||
&& (this.settings?.model !== targetId || (this.settings.effort ?? null) !== (effort ?? null))) return;
|
||||
this.acceptSettings({ model: targetId, effort, serviceTier: null });
|
||||
if (!entering) {
|
||||
const path = this.returnPath();
|
||||
if (path) await rm(path, { force: true }).catch(error => logger.debug('[Codex] Failed to remove Reserve return record', error));
|
||||
}
|
||||
this.publishUsage();
|
||||
this.notice(entering
|
||||
? 'Ordinary Codex usage is exhausted. Switched to GPT-5.6 Luna using Luna Reserve.'
|
||||
: `Ordinary Codex usage is available again. Switched back to ${targetId}.`);
|
||||
} catch (error) {
|
||||
logger.debug('[Codex] Reserve reconciliation unavailable', error);
|
||||
if (applying) {
|
||||
this.supported = false;
|
||||
this.notice('Codex could not apply the Reserve transition. Task settings are unchanged; no message was retried.');
|
||||
}
|
||||
this.publishUsage();
|
||||
}
|
||||
}
|
||||
|
||||
turnMode(mode: EnhancedMode): EnhancedMode {
|
||||
if (this.settings?.model !== RESERVE_MODEL) return mode;
|
||||
const target = this.models.find(model => (model.model ?? model.id) === RESERVE_MODEL);
|
||||
const effort = target?.supportedReasoningEfforts?.some(item => item.reasoningEffort === mode.modelReasoningEffort)
|
||||
? mode.modelReasoningEffort : this.settings.effort ?? undefined;
|
||||
return { ...mode, model: RESERVE_MODEL, modelReasoningEffort: effort, serviceTier: 'standard' };
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user