wushenghua/hapi
1
0
Fork 0
mirror of https://github.com/wu736139669/hapi.git synced 2026-11-08 01:43:07 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(codex): distinguish compaction waits from unknown delivery (#2115)

Browse Source
This commit is contained in:
Ananovo
2026-10-25 15:34:52 +08:00
committed by GitHub
parent 3eadf65f21
commit cb2ff05c2a
11 changed files with 194 additions and 13 deletions
Download Patch File Download Diff File Expand all files Collapse all files
cli/src/codex/shared/queue.ts
+5 -3
View File
@@ -80,8 +80,8 @@ export class SharedCodexQueue {
} while (cursor);
return items;
}
reconcile(): Promise<void> { return this.serial(() => this.reconcileNow()); }
private async reconcileNow(): Promise<void> {
reconcile(): Promise<string[]> { return this.serial(() => this.reconcileNow()); }
private async reconcileNow(): Promise<string[]> {
const present = new Set<string>();
for (const item of await this.list()) {
const id = item.clientUserMessageId; present.add(id);
@@ -94,13 +94,15 @@ export class SharedCodexQueue {
}
// Absence could mean consumed, removed, or lost during engine restart.
// Only an item event/history or a successful delete can decide which.
const dequeued: string[] = [];
for (const [id, entry] of Object.entries(this.entries)) {
if (entry.state === 'queued' && !present.has(id)) entry.state = 'unknown';
if (entry.state === 'queued' && !present.has(id)) { entry.state = 'unknown'; dequeued.push(id); }
}
await this.save();
const unknown = Object.entries(this.entries).filter(([, entry]) => entry.state === 'unknown').map(([id]) => id);
if (unknown.length) this.uncertain(unknown);
await this.publishReleased();
return dequeued;
}
private async publishReleased(): Promise<void> {
cli/src/codex/shared/root.test.ts
+88 -2
View File
@@ -61,6 +61,8 @@ async function fixture(opts?: { hubArchived?: boolean; end?: RootHost['end'] })
const updateState = vi.fn((fn: (value: AgentState) => AgentState) => { state = fn(state); });
const rpc = new Map<string, (raw: unknown) => Promise<unknown>>();
const send = vi.fn();
const consumed = vi.fn();
const uncertain = vi.fn();
const hubArchivedListeners: Array<() => void> = [];
const session = {
sessionId: 'sid', getMetadata: () => metadata,
@@ -74,7 +76,7 @@ async function fixture(opts?: { hubArchived?: boolean; end?: RootHost['end'] })
},
rpcHandlerManager: { registerHandler: (name: string, handler: (raw: unknown) => Promise<unknown>) => rpc.set(name, handler) },
sendSessionEvent() {}, sendAgentMessage: send, emitSessionReady() {},
sendUserMessage() {}, emitMessagesConsumed() {}, emitSteerIndeterminate() {}, syncNativeQueuedMessage() {},
sendUserMessage() {}, emitMessagesConsumed: consumed, emitSteerIndeterminate: uncertain, syncNativeQueuedMessage() {},
sendSessionDeath() {}, async flush() {}, close() {}
} as unknown as ApiSessionClient;
const end = opts?.end ?? (async () => { throw new Error('Unexpected root archive'); });
@@ -94,7 +96,8 @@ async function fixture(opts?: { hubArchived?: boolean; end?: RootHost['end'] })
abandoned(): void;
};
return {
root, native, rpc, send, metadata: () => metadata, state: () => state, updateState,
root, native, rpc, send, consumed, uncertain, metadata: () => metadata, state: () => state, updateState,
drain: () => (root as unknown as { notifications: Promise<void> }).notifications,
reconnect: () => reconnect?.(),
emitHubArchived: () => { for (const listener of hubArchivedListeners) listener(); },
hubArchivedListenerCount: () => hubArchivedListeners.length,
@@ -102,6 +105,89 @@ async function fixture(opts?: { hubArchived?: boolean; end?: RootHost['end'] })
};
}
async function startCompaction(f: Awaited<ReturnType<typeof fixture>>, id = 'input', turnId = 'compacting-turn') {
f.native.queue = [{ id: `native-${id}`, clientUserMessageId: id, input: [{ type: 'text', text: id }] }];
await f.root.refresh();
f.native.queue = [];
// Native removes input before announcing the turn and before recording userMessage.
f.native.notify('thread/queue/changed', { threadId: 'thread' });
await vi.waitFor(() => expect(f.uncertain).toHaveBeenCalledWith(expect.arrayContaining([id])));
const turn = { id: turnId, status: 'inProgress', items: [] };
f.native.thread.turns.push(turn);
f.native.notify('turn/started', { threadId: 'thread', turn });
f.native.notify('item/started', { threadId: 'thread', turnId, item: { id: 'compact', type: 'contextCompaction' } });
return turn;
}
describe('compaction delivery confirmation', () => {
it('does not turn uncertainty discovered by a cold snapshot into live compaction evidence', async () => {
const f = await fixture();
f.native.queue = [{ id: 'old-native', clientUserMessageId: 'old', input: [] }];
await f.root.refresh();
f.native.queue = [];
await f.root.refresh();
expect(f.uncertain).toHaveBeenCalledWith(['old']);
f.native.thread.turns.push({ id: 'later', status: 'inProgress', items: [] });
f.native.notify('turn/started', { threadId: 'thread', turn: { id: 'later' } });
f.native.notify('item/started', { threadId: 'thread', turnId: 'later', item: { id: 'c', type: 'contextCompaction' } });
await f.drain();
expect(f.state()).toMatchObject({ codexCompactionPendingMessageIds: [] });
});
it('reports only freshly dequeued input as waiting, without acknowledging or replaying it', async () => {
const f = await fixture();
const request = vi.spyOn(f.root.client, 'request');
const turn = await startCompaction(f);
await vi.waitFor(() => expect(f.state()).toMatchObject({ codexCompactionPendingMessageIds: ['input'] }));
expect(f.consumed).not.toHaveBeenCalled();
expect(request.mock.calls.some(([method]) => method === 'thread/queue/add')).toBe(false);
f.native.notify('item/started', { threadId: 'thread', turnId: turn.id,
item: { id: 'user', type: 'userMessage', clientId: 'input', content: [] } });
await vi.waitFor(() => expect(f.consumed).toHaveBeenCalledWith(['input'], { steered: undefined }));
expect(f.state()).toMatchObject({ codexCompactionPendingMessageIds: [] });
});
it.each(['completed', 'interrupted', 'failed'])('restores uncertainty when the turn is %s without consumption', async status => {
const f = await fixture();
const turn = await startCompaction(f);
await vi.waitFor(() => expect(f.state()).toMatchObject({ codexCompactionPendingMessageIds: ['input'] }));
turn.status = status;
f.native.notify('turn/completed', { threadId: 'thread', turn });
await vi.waitFor(() => expect(f.state()).toMatchObject({ codexCompactionPendingMessageIds: [] }));
expect(f.consumed).not.toHaveBeenCalled();
await startCompaction(f, 'next', 'next-turn');
await vi.waitFor(() => expect(f.state()).toMatchObject({ codexCompactionPendingMessageIds: ['next'] }));
});
it('withdraws waiting at compaction completion and ignores child and stale turn events', async () => {
const f = await fixture();
const turn = await startCompaction(f);
await vi.waitFor(() => expect(f.state()).toMatchObject({ codexCompactionPendingMessageIds: ['input'] }));
for (const [threadId, turnId] of [['child', turn.id], ['thread', 'stale']]) {
f.native.notify('item/completed', { threadId, turnId, item: { id: 'other', type: 'contextCompaction' } });
}
await f.drain();
expect(f.state()).toMatchObject({ codexCompactionPendingMessageIds: ['input'] });
f.native.notify('item/completed', { threadId: 'thread', turnId: turn.id, item: { id: 'compact', type: 'contextCompaction' } });
await vi.waitFor(() => expect(f.state()).toMatchObject({ codexCompactionPendingMessageIds: [] }));
expect(f.consumed).not.toHaveBeenCalled();
});
it('does not keep waiting after native disconnect or stopping', async () => {
const f = await fixture();
await startCompaction(f);
await vi.waitFor(() => expect(f.state()).toMatchObject({ codexCompactionPendingMessageIds: ['input'] }));
f.native.initialized = false;
f.native.abandoned();
expect(f.state()).toMatchObject({ codexCompactionPendingMessageIds: [] });
await vi.waitFor(() => expect(f.native.initialized).toBe(true));
await f.root.refresh();
expect(f.state()).toMatchObject({ codexCompactionPendingMessageIds: [] });
f.root.stopAccepting();
expect(f.state()).toMatchObject({ codexCompactionPendingMessageIds: [] });
});
});
async function completePlan(f: Awaited<ReturnType<typeof fixture>>, status = 'completed') {
await f.root.applySettings({ collaborationMode: 'plan' });
const turn = { id: 'plan-turn', status: 'inProgress', items: [{ id: 'plan-item', type: 'plan', text: '# Implement me' }] };
cli/src/codex/shared/root.ts
+44 -5
View File
@@ -54,6 +54,11 @@ export class SharedCodexRoot {
private work: Promise<unknown> = Promise.resolve();
private notifications = Promise.resolve();
private currentTurn: string | undefined;
// Ephemeral observations only: dequeue/compaction never prove acceptance.
private readonly dequeuedInputs = new Set<string>();
private deliveryTurn?: string;
private compactingTurn?: string;
private publishedCompactionIds?: string;
private latestTurn?: { id: string; status: string; planId?: string };
private publishedPlanId: string | null | undefined;
private submittingPlan = false;
@@ -137,6 +142,7 @@ export class SharedCodexRoot {
this.projection?.reset();
this.steeringActive = undefined;
this.publishedPlanId = undefined;
this.publishedCompactionIds = undefined;
void this.refresh().then(() => this.refreshChildren(false)).then(() => this.queue.replay())
.catch(error => logger.debug('[Codex shared] hub resync', error));
});
@@ -186,7 +192,11 @@ export class SharedCodexRoot {
if (this.threadId && this.threadId !== threadId) throw new Error('Cannot retarget a shared HAPI session');
this.threadId = threadId;
this.queue = new SharedCodexQueue(this.client, threadId, join(this.host.directory, `${this.session.sessionId}.queue.json`),
(ids, steered) => this.session.emitMessagesConsumed(ids, { steered }), ids => this.session.emitSteerIndeterminate(ids),
(ids, steered) => {
this.session.emitMessagesConsumed(ids, { steered });
for (const id of ids) this.dequeuedInputs.delete(id);
this.publishCompaction();
}, ids => this.session.emitSteerIndeterminate(ids),
(id, input) => this.session.syncNativeQueuedMessage(id, input === null ? null : inputText(input)),
ids => this.session.setSteerDeliveryState(ids, 'queued'));
await this.queue.load();
@@ -197,7 +207,7 @@ export class SharedCodexRoot {
// advertise a destructive operation we cannot make atomic yet.
conversationHistory: { forkCurrent: true, forkAtMessage: true, rewindToMessage: false }
} }));
this.session.updateAgentState(state => ({ ...state, controlledByUser: false, startingMode: undefined, codexPlanProposalId: null, requests: {},
this.session.updateAgentState(state => ({ ...state, controlledByUser: false, startingMode: undefined, codexPlanProposalId: null, codexCompactionPendingMessageIds: [], requests: {},
completedRequests: { ...state.completedRequests, ...Object.fromEntries(Object.entries(state.requests ?? {}).map(([id, request]) =>
[id, { ...request, completedAt: Date.now(), status: 'canceled' as const }])) }
}));
@@ -213,11 +223,21 @@ export class SharedCodexRoot {
this.heartbeat = setInterval(() => this.alive(), 2_000); this.alive(); this.session.emitSessionReady();
}
private publishSteering(): void {
this.publishCompaction();
const active = Boolean(this.currentTurn) && !this.stopping && !this.closed && !this.reconnecting && this.client.isInitialized();
if (this.steeringActive === active) return;
this.steeringActive = active;
this.session.updateAgentState(state => ({ ...state, steeringActive: active }));
}
private publishCompaction(): void {
const active = this.compactingTurn && this.compactingTurn === this.currentTurn
&& !this.closed && !this.stopping && !this.reconnecting && this.client.isInitialized();
const ids = active ? [...this.dequeuedInputs].filter(id => this.queue?.state(id) === 'unknown').sort() : [];
const key = JSON.stringify(ids);
if (key === this.publishedCompactionIds) return;
this.publishedCompactionIds = key;
this.session.updateAgentState(state => ({ ...state, codexCompactionPendingMessageIds: ids }));
}
private alive(): void {
this.publishSteering();
this.publishPlan();
@@ -273,8 +293,25 @@ export class SharedCodexRoot {
const name = p.threadName ?? undefined; this.session.updateMetadata(metadata => ({ ...metadata, name }));
}
if (method === 'thread/archived') { await this.host.end(this, false); return; }
if (method === 'thread/queue/changed') await this.queue.reconcile();
await this.projection.notification(method, params, modelAtReceipt); this.alive();
// Process these in notification order: queue removal can precede turn/started.
// Older unknown input must not acquire a later turn's compaction status.
if (method === 'turn/started') {
const id = string(record(p.turn).id);
if (this.deliveryTurn && this.deliveryTurn !== id) this.dequeuedInputs.clear();
this.deliveryTurn = id;
this.compactingTurn = undefined;
}
if (method === 'turn/completed' && record(p.turn).id === this.deliveryTurn) {
this.dequeuedInputs.clear(); this.deliveryTurn = undefined; this.compactingTurn = undefined;
}
if (p.turnId === this.currentTurn && record(p.item).type === 'contextCompaction') {
if (method === 'item/started') this.compactingTurn = string(p.turnId);
if (method === 'item/completed') { this.compactingTurn = undefined; this.dequeuedInputs.clear(); }
}
if (method === 'thread/queue/changed') {
for (const id of await this.queue.reconcile()) this.dequeuedInputs.add(id);
}
await this.projection.notification(method, params, modelAtReceipt); this.publishCompaction(); this.alive();
}
async readThread(threadId = this.threadId): Promise<Record<string, unknown>> {
let thread = record(record(await this.client.request('thread/read', { threadId, includeTurns: false })).thread);
@@ -342,7 +379,9 @@ export class SharedCodexRoot {
}
private async reconnect(): Promise<void> {
if (this.closed || this.stopping || this.reconnecting || !this.threadId) return;
this.reconnecting = true; this.publishSteering(); this.publishPlan(); this.permissions.close();
this.reconnecting = true;
this.dequeuedInputs.clear(); this.deliveryTurn = undefined; this.compactingTurn = undefined;
this.publishSteering(); this.publishPlan(); this.permissions.close();
try {
while (!this.closed && !this.stopping) {
try {