skills/pbakaus_impeccable
1
0
Fork 0
mirror of https://github.com/pbakaus/impeccable.git synced 2026-09-12 06:06:37 +03:00
Code Issues Packages Projects Releases Wiki Activity

Detach canceled Live generation tails

Browse Source 
AI-assisted: Codex
This commit is contained in:
Paul Bakaus
2026-07-13 10:08:27 -07:00
parent 2008381e91
commit db63d08168
3 changed files with 118 additions and 21 deletions
Download Patch File Download Diff File Expand all files Collapse all files
scripts/benchmark-live.mjs
+3
View File
@@ -204,6 +204,9 @@ try {
if (args.action) await selectAction(session.page, String(args.action));
await resetBrowserTimingProbe(session.page, `${iteration}-followup`);
const nextFirstVariant = waitForFirstVariant(session.page);
// Keep teardown from surfacing this background waiter as an unhandled
// rejection if a preceding follow-up action is the real failure.
void nextFirstVariant.catch(() => {});
await clickGo(session.page);
await waitForBrowserGeneratePost(session.page);
run.acceptToNextGoDispatchMs = roundMs(performance.now() - acceptStartedAt);
skill/scripts/live/codex-worker-supervisor.mjs
+70 -20
View File
@@ -74,6 +74,7 @@ export class CodexLiveWorkerSupervisor {
this.canceled = new Set();
this.queuedGenerationIds = new Set();
this.thread = null;
this.threadReady = Promise.resolve(null);
this.model = null;
this.liveSpec = '';
}
@@ -104,16 +105,9 @@ export class CodexLiveWorkerSupervisor {
}
}
if (!this.thread) {
this.thread = await this.client.startDedicatedThread({
model: this.model.model || this.model.id,
cwd: this.cwd,
approvalPolicy: 'never',
sandbox: 'read-only',
ephemeral: false,
serviceName: 'impeccable_live_codex_worker',
baseInstructions: buildCodexWorkerInstructions(this.liveSpec),
});
this.thread = await this.startWorkerThread();
}
this.threadReady = Promise.resolve(this.thread);
this.writeState('ready');
return this.status();
}
@@ -134,10 +128,12 @@ export class CodexLiveWorkerSupervisor {
}
if (event.type === 'accept' || event.type === 'discard') {
this.canceled.add(event.id);
const replaceBusyThread = this.active?.eventId === event.id;
// Cancellation fences publication synchronously. Do not make the
// deterministic Accept/Discard path wait on a slow app-server
// interrupt round trip before it can update source and reply.
void this.cancelActive(event.type, event.id);
if (replaceBusyThread) this.rotateWorkerThread(event.type);
const handled = await this.handleAccept(event, this.base, this.token);
if (event.type === 'accept' && handled?._acceptResult?.carbonize === true) {
await this.postCleanup(this.base, this.token, {
@@ -173,12 +169,14 @@ export class CodexLiveWorkerSupervisor {
}
async processGeneration(event) {
if (this.isCanceled(event.id)) return;
await this.threadReady;
if (this.isCanceled(event.id)) return;
if (!event.scaffold?.file) event.scaffold = runDeterministicScaffold(event, {
cwd: this.cwd,
scriptsDir: this.scriptsDir,
});
this.active = { eventId: event.id, turnId: null };
this.active = { eventId: event.id, turnId: null, threadId: this.thread.id };
this.writeState('working', { eventId: event.id });
try {
const expectedVariants = Number(event.count || 1);
@@ -202,11 +200,52 @@ export class CodexLiveWorkerSupervisor {
file: event.scaffold.file,
});
} finally {
this.active = null;
this.writeState('ready');
if (this.active?.eventId === event.id) {
this.active = null;
this.writeState('ready');
}
}
}
startWorkerThread() {
return this.client.startDedicatedThread({
model: this.model.model || this.model.id,
cwd: this.cwd,
approvalPolicy: 'never',
sandbox: 'read-only',
ephemeral: false,
serviceName: 'impeccable_live_codex_worker',
baseInstructions: buildCodexWorkerInstructions(this.liveSpec),
});
}
rotateWorkerThread(reason) {
const priorThread = this.thread;
const drainingQueue = this.queue;
this.queue = Promise.resolve();
this.thread = null;
this.threadReady = this.startWorkerThread().then((thread) => {
this.thread = thread;
this.writeState('ready', {
rotatedAt: new Date().toISOString(),
rotationReason: reason,
});
return thread;
});
void this.threadReady.catch((error) => {
this.writeState('error', { error: error.message, rotationReason: reason });
this.log(`replacement worker thread failed: ${error.message}`);
});
if (priorThread) {
void drainingQueue.finally(async () => {
await this.client.archiveThread(priorThread.id).catch((error) => {
this.log(`retired worker thread archive failed: ${error.message}`);
});
});
}
return this.threadReady;
}
async runGenerationPhase(event, phase, arrivedVariants) {
if (this.isCanceled(event.id)) return;
const phaseStartedAt = Date.now();
@@ -277,22 +316,25 @@ export class CodexLiveWorkerSupervisor {
if (publicationPromise === pendingPublication) publicationPromise = null;
}
};
if (this.isCanceled(event.id)) return;
const result = await this.runTurnWithReconnect({
input,
outputSchema: CODEX_WORKER_OUTPUT_SCHEMA,
onAgentMessage: publishCandidate,
eventId: event.id,
});
if (this.isCanceled(event.id)) return;
if (!publishedFromMessage) await publishCandidate(result.answer);
if (!publishedFromMessage) throw earlyCandidateError || supervisorError('worker_output_not_published');
}
async runTurnWithReconnect({ input, outputSchema, onAgentMessage }) {
async runTurnWithReconnect({ input, outputSchema, onAgentMessage, eventId = this.active?.eventId }) {
let firstError;
for (let attempt = 0; attempt < 2; attempt += 1) {
try {
const threadId = this.thread.id;
const turn = await this.client.startTurn({
threadId: this.thread.id,
threadId,
input,
cwd: this.cwd,
model: this.model.model || this.model.id,
@@ -303,16 +345,16 @@ export class CodexLiveWorkerSupervisor {
outputSchema,
onAgentMessage,
onStarted: (turnId) => {
if (!this.active) return;
this.active.turnId = turnId;
if (this.isCanceled(this.active.eventId)) {
this.client.interruptTurn(this.thread.id, turnId).catch(() => {});
if (this.active?.eventId === eventId) this.active.turnId = turnId;
if (eventId && this.isCanceled(eventId)) {
this.client.interruptTurn(threadId, turnId).catch(() => {});
}
},
});
return { ...turn, answer: turn.message };
} catch (error) {
if (!firstError) firstError = error;
if (eventId && this.isCanceled(eventId)) throw error;
if (attempt > 0 || error.code === 'TURN_INTERRUPTED') throw error;
this.log(`app-server turn failed; reconnecting once: ${error.message}`);
await this.reconnect();
@@ -339,8 +381,9 @@ export class CodexLiveWorkerSupervisor {
if (!this.active) return;
if (eventId && this.active.eventId !== eventId) return;
this.canceled.add(this.active.eventId);
if (this.active.turnId) {
await this.client.interruptTurn(this.thread.id, this.active.turnId).catch(() => {});
const threadId = this.active.threadId || this.thread?.id;
if (threadId && this.active.turnId) {
await this.client.interruptTurn(threadId, this.active.turnId).catch(() => {});
}
this.log(`interrupted ${this.active.eventId}: ${reason}`);
}
@@ -363,6 +406,13 @@ export class CodexLiveWorkerSupervisor {
async shutdown({ archive = false } = {}) {
this.running = false;
await this.cancelActive('shutdown');
await Promise.race([
this.threadReady.catch(() => null),
new Promise((resolve) => {
const timer = setTimeout(resolve, 1_000);
timer.unref?.();
}),
]);
let archived = false;
if (archive && this.thread) {
try {
tests/live-codex-worker-supervisor.test.mjs
+45 -1
View File
@@ -101,6 +101,7 @@ describe('Codex Live worker supervisor ownership and lifecycle', () => {
client,
});
supervisor.thread = { id: 'live-worker-thread' };
supervisor.model = client.models[0];
supervisor.active = { eventId: 'generation-1', turnId: 'turn-1' };
await supervisor.cancelActive('accept', 'generation-1');
assert.deepEqual(client.calls.interruptTurn, [{ threadId: 'live-worker-thread', turnId: 'turn-1' }]);
@@ -137,7 +138,8 @@ describe('Codex Live worker supervisor ownership and lifecycle', () => {
},
});
supervisor.thread = { id: 'live-worker-thread' };
supervisor.active = { eventId: 'generation-1', turnId: 'turn-1' };
supervisor.model = client.models[0];
supervisor.active = { eventId: 'generation-1', turnId: 'turn-1', threadId: 'live-worker-thread' };
await supervisor.run();
@@ -145,6 +147,46 @@ describe('Codex Live worker supervisor ownership and lifecycle', () => {
assert.equal(client.calls.interruptTurn.length >= 1, true);
});
it('rotates a busy thread so the next generation does not wait for the canceled tail', async () => {
const cwd = mkdtempSync(path.join(tmpdir(), 'codex-supervisor-tail-rotation-'));
const client = fakeClient();
client.startDedicatedThread = async (params) => {
client.calls.startDedicatedThread.push(params);
await new Promise((resolve) => setImmediate(resolve));
return { id: 'replacement-live-thread' };
};
const supervisor = createSupervisor({ cwd, statePath: path.join(cwd, 'state.json'), client });
supervisor.model = client.models[0];
supervisor.thread = { id: 'draining-live-thread' };
let releaseDrainingQueue;
let drainingQueueFinished = false;
supervisor.queue = new Promise((resolve) => {
releaseDrainingQueue = () => {
drainingQueueFinished = true;
resolve();
};
});
let observedThread = null;
supervisor.runGenerationPhase = async () => {
observedThread = supervisor.thread.id;
};
supervisor.reply = async () => {};
supervisor.rotateWorkerThread('accept');
await supervisor.processGeneration({
type: 'generate',
id: 'next-generation',
count: 1,
scaffold: { file: 'src/App.jsx' },
});
assert.equal(observedThread, 'replacement-live-thread');
assert.equal(drainingQueueFinished, false, 'the next generation must not join the canceled tail queue');
releaseDrainingQueue();
await new Promise((resolve) => setImmediate(resolve));
assert.deepEqual(client.calls.archiveThread, [{ threadId: 'draining-live-thread' }]);
});
it('interrupts a canceled turn whose id arrives after Accept', async () => {
const cwd = mkdtempSync(path.join(tmpdir(), 'codex-supervisor-late-turn-'));
const client = fakeClient();
@@ -260,6 +302,8 @@ describe('Codex Live worker supervisor ownership and lifecycle', () => {
scriptsDir: path.join(cwd, 'skill/scripts'),
reply: async (_base, _token, value) => { replies.push(value); },
});
supervisor.thread = { id: 'live-worker-thread' };
supervisor.threadReady = Promise.resolve(supervisor.thread);
supervisor.runGenerationPhase = async (_event, phase, arrivedVariants) => {
phases.push({ phase, arrivedVariants });
};