fix: add Code Queue steer confirmation
This commit is contained in:
Codex
@@ -136,6 +136,7 @@ import {
|
||||
} from "./oa-events";
|
||||
import { collectRuntimePreflight, runtimePreflightJson } from "./runtime-preflight";
|
||||
import { collectSkillAvailability, collectSkillSyncPreflight, skillAvailabilityJson, skillSyncPreflightJson } from "./skill-availability";
|
||||
import { createSteerId, findSteerTraceConfirmation, normalizeSteerId, normalizeSteerPromptText, steerDuplicateDecision, steerPromptHash, steerTraceConfirmationJson, steerTraceText } from "./steer-confirmation";
|
||||
import { configureSelfTests, runJudgeInfraSelfTest, runQueueClaimMoveSelfTest, runQueueOrderingSelfTest, runReferenceInjectionSelfTest, runStaleActiveRecoverySelfTest, runTracePortSelfTest, runTraceSummaryContractSelfTest } from "./self-tests";
|
||||
import {
|
||||
codexToolLifecycleStartedBeforeIn,
|
||||
@@ -918,10 +919,6 @@ function taskQueueEnteredAt(task: QueueTask): string {
|
||||
return entryEvents[0] ?? taskTimestamp(task.createdAt) ?? taskTimestamp(task.updatedAt) ?? nowIso();
|
||||
}
|
||||
|
||||
function normalizeSteerPromptText(text: string): string {
|
||||
return text.replace(/^\s*\[steer\]\s*/u, "").trimEnd();
|
||||
}
|
||||
|
||||
function outputPromptHistory(task: QueueTask): PromptHistoryItem[] {
|
||||
return task.output
|
||||
.filter((item) => item.channel === "user" && item.method === "turn/steer")
|
||||
@@ -930,6 +927,7 @@ function outputPromptHistory(task: QueueTask): PromptHistoryItem[] {
|
||||
at: item.at,
|
||||
method: "turn/steer" as const,
|
||||
text: normalizeSteerPromptText(item.text),
|
||||
...(normalizeSteerId(item.itemId) === null ? {} : { steerId: normalizeSteerId(item.itemId) as string }),
|
||||
}))
|
||||
.filter((item) => item.text.trim().length > 0);
|
||||
}
|
||||
@@ -940,7 +938,8 @@ function mergePromptHistory(items: PromptHistoryItem[]): PromptHistoryItem[] {
|
||||
const seq = Number(item.seq);
|
||||
const text = normalizeSteerPromptText(String(item.text || ""));
|
||||
if (!Number.isFinite(seq) || text.trim().length === 0) continue;
|
||||
byKey.set(`${seq}:${item.method}`, { seq, at: item.at || nowIso(), method: "turn/steer", text });
|
||||
const steerId = normalizeSteerId((item as PromptHistoryItem & { steerId?: unknown }).steerId);
|
||||
byKey.set(`${seq}:${item.method}`, { seq, at: item.at || nowIso(), method: "turn/steer", text, ...(steerId === null ? {} : { steerId }) });
|
||||
}
|
||||
return Array.from(byKey.values()).sort((left, right) => left.seq - right.seq);
|
||||
}
|
||||
@@ -2646,13 +2645,14 @@ function createTask(request: QueueTaskRequest): QueueTask {
|
||||
};
|
||||
}
|
||||
|
||||
function appendPromptHistory(task: QueueTask, output: LiveOutput | null, method: PromptHistoryItem["method"], text: string): void {
|
||||
function appendPromptHistory(task: QueueTask, output: LiveOutput | null, method: PromptHistoryItem["method"], text: string, options: { steerId?: string } = {}): void {
|
||||
if (output === null) return;
|
||||
task.promptHistory = mergePromptHistory([...(Array.isArray(task.promptHistory) ? task.promptHistory : []), {
|
||||
seq: output.seq,
|
||||
at: output.at,
|
||||
method,
|
||||
text,
|
||||
...(options.steerId === undefined ? {} : { steerId: options.steerId }),
|
||||
}]);
|
||||
markTaskDirty(task.id);
|
||||
schedulePersistState();
|
||||
@@ -4812,17 +4812,65 @@ async function createTasks(req: Request): Promise<Response> {
|
||||
async function steerTask(task: QueueTask, req: Request): Promise<Response> {
|
||||
if (!serviceRoleAllowsScheduler(config.serviceRole)) return schedulerOnlyRejectResponse(req.method, `/api/tasks/${task.id}/steer`);
|
||||
const body = await readJson(req);
|
||||
const prompt = typeof (body as Record<string, unknown>).prompt === "string" ? String((body as Record<string, unknown>).prompt) : "";
|
||||
const bodyRecord = typeof body === "object" && body !== null && !Array.isArray(body) ? body as Record<string, unknown> : {};
|
||||
const prompt = typeof bodyRecord.prompt === "string" ? String(bodyRecord.prompt) : "";
|
||||
if (prompt.trim().length === 0) return jsonResponse({ ok: false, error: "prompt is required" }, 400);
|
||||
const steerId = normalizeSteerId(bodyRecord.steerId) ?? createSteerId(task.id, prompt);
|
||||
const duplicateDecision = steerDuplicateDecision(task, steerId, prompt, taskFullOutput(task));
|
||||
if (duplicateDecision.duplicate) {
|
||||
await flushDirtyTasksToDatabase(true);
|
||||
return jsonResponse({
|
||||
ok: true,
|
||||
accepted: true,
|
||||
duplicateSuppressed: true,
|
||||
deliveryState: "accepted",
|
||||
steerId,
|
||||
traceConfirmation: steerTraceConfirmationJson(duplicateDecision.confirmation),
|
||||
task: taskForResponse(task),
|
||||
queue: await queueSummaryForResponse(),
|
||||
});
|
||||
}
|
||||
if (duplicateDecision.conflict) {
|
||||
return jsonResponse({
|
||||
ok: false,
|
||||
error: "steerId already exists with a different prompt hash",
|
||||
accepted: false,
|
||||
deliveryState: "not_accepted",
|
||||
steerId,
|
||||
existingPromptHash: duplicateDecision.existingPromptHash,
|
||||
requestedPromptHash: duplicateDecision.requestedPromptHash,
|
||||
traceConfirmation: steerTraceConfirmationJson(duplicateDecision.confirmation),
|
||||
task: taskForResponse(task),
|
||||
}, 409);
|
||||
}
|
||||
const activeRun = activeRunForTask(task);
|
||||
if (activeRun === null || activeRun.threadId === null || activeRun.turnId === null || typeof activeRun.app.steer !== "function") {
|
||||
return jsonResponse({ ok: false, error: "task does not have an active steerable turn", task: taskForResponse(task) }, 409);
|
||||
return jsonResponse({
|
||||
ok: false,
|
||||
error: "task does not have an active steerable turn",
|
||||
accepted: false,
|
||||
deliveryState: "not_accepted",
|
||||
steerId,
|
||||
traceConfirmation: steerTraceConfirmationJson(duplicateDecision.confirmation),
|
||||
task: taskForResponse(task),
|
||||
}, 409);
|
||||
}
|
||||
const output = appendOutput(task, "user", `\n[steer] ${prompt}\n`, "turn/steer");
|
||||
appendPromptHistory(task, output, "turn/steer", prompt);
|
||||
const output = appendOutput(task, "user", steerTraceText(steerId, prompt), "turn/steer", steerId);
|
||||
appendPromptHistory(task, output, "turn/steer", prompt, { steerId });
|
||||
await activeRun.app.steer(activeRun.threadId, activeRun.turnId, prompt);
|
||||
await flushDirtyTasksToDatabase(true);
|
||||
return jsonResponse({ ok: true, task: taskForResponse(task), queue: await queueSummaryForResponse() });
|
||||
const traceConfirmation = findSteerTraceConfirmation(task, steerId, taskFullOutput(task));
|
||||
return jsonResponse({
|
||||
ok: true,
|
||||
accepted: true,
|
||||
duplicateSuppressed: false,
|
||||
deliveryState: traceConfirmation.found ? "accepted" : "accepted_response_timeout",
|
||||
steerId,
|
||||
promptHash: steerPromptHash(prompt),
|
||||
traceConfirmation: steerTraceConfirmationJson(traceConfirmation),
|
||||
task: taskForResponse(task),
|
||||
queue: await queueSummaryForResponse(),
|
||||
});
|
||||
}
|
||||
|
||||
async function editQueuedTaskPrompt(task: QueueTask, req: Request): Promise<Response> {
|
||||
@@ -5726,6 +5774,17 @@ async function route(req: Request): Promise<Response> {
|
||||
if (task === null) return jsonResponse({ ok: false, error: "task not found" }, 404);
|
||||
return await taskTraceStepDetailResponse(task, url);
|
||||
}
|
||||
const steerConfirmationMatch = url.pathname.match(/^\/api\/tasks\/([^/]+)\/steer-confirmation$/u);
|
||||
if (steerConfirmationMatch !== null && req.method === "GET") {
|
||||
const task = await findTaskForRead(decodeURIComponent(steerConfirmationMatch[1] ?? ""));
|
||||
if (task === null) return jsonResponse({ ok: false, error: "task not found" }, 404);
|
||||
const steerId = normalizeSteerId(url.searchParams.get("steerId"));
|
||||
if (steerId === null) return jsonResponse({ ok: false, error: "steerId is required" }, 400);
|
||||
return jsonResponse({
|
||||
ok: true,
|
||||
confirmation: steerTraceConfirmationJson(findSteerTraceConfirmation(task, steerId, taskFullOutput(task))),
|
||||
});
|
||||
}
|
||||
const judgeTaskMatch = url.pathname.match(/^\/api\/tasks\/([^/]+)\/judge$/u);
|
||||
if (judgeTaskMatch !== null && (req.method === "GET" || req.method === "POST")) {
|
||||
const task = await findTaskForRead(decodeURIComponent(judgeTaskMatch[1] ?? ""));
|
||||
|
||||
@@ -0,0 +1,157 @@
|
||||
import { createHash } from "node:crypto";
|
||||
import type { JsonValue, LiveOutput, PromptHistoryItem, QueueTask } from "./types";
|
||||
|
||||
const steerIdPattern = /^[A-Za-z0-9._:-]{8,128}$/u;
|
||||
|
||||
export type SteerDeliveryState = "accepted" | "not_accepted" | "accepted_response_timeout" | "unknown";
|
||||
|
||||
export interface SteerTraceMatch {
|
||||
seq: number;
|
||||
at: string;
|
||||
method: "turn/steer";
|
||||
steerId: string;
|
||||
promptChars: number;
|
||||
promptHash: string;
|
||||
promptOmitted: true;
|
||||
source: "promptHistory" | "output";
|
||||
}
|
||||
|
||||
export interface SteerTraceConfirmation {
|
||||
taskId: string;
|
||||
steerId: string;
|
||||
found: boolean;
|
||||
accepted: boolean;
|
||||
deliveryState: SteerDeliveryState;
|
||||
trace: SteerTraceMatch | null;
|
||||
matches: SteerTraceMatch[];
|
||||
duplicateSuppressionKey: string;
|
||||
}
|
||||
|
||||
export interface SteerDuplicateDecision {
|
||||
duplicate: boolean;
|
||||
conflict: boolean;
|
||||
confirmation: SteerTraceConfirmation;
|
||||
existingPromptHash: string | null;
|
||||
requestedPromptHash: string;
|
||||
}
|
||||
|
||||
export function normalizeSteerId(value: unknown): string | null {
|
||||
if (typeof value !== "string") return null;
|
||||
const text = value.trim();
|
||||
if (!steerIdPattern.test(text)) return null;
|
||||
return text;
|
||||
}
|
||||
|
||||
export function steerPromptHash(prompt: string): string {
|
||||
return createHash("sha256").update(prompt, "utf8").digest("hex");
|
||||
}
|
||||
|
||||
export function createSteerId(taskId: string, prompt: string): string {
|
||||
const hash = createHash("sha256")
|
||||
.update("unidesk-code-queue-steer:v1", "utf8")
|
||||
.update("\0", "utf8")
|
||||
.update(taskId, "utf8")
|
||||
.update("\0", "utf8")
|
||||
.update(prompt, "utf8")
|
||||
.digest("hex")
|
||||
.slice(0, 24);
|
||||
return `steer_${hash}`;
|
||||
}
|
||||
|
||||
export function steerTraceText(steerId: string, prompt: string): string {
|
||||
return `\n[steer id=${steerId}] ${prompt}\n`;
|
||||
}
|
||||
|
||||
export function normalizeSteerPromptText(text: string): string {
|
||||
return text.replace(/^\s*\[steer(?:\s+[^\]]*)?\]\s*/u, "").trimEnd();
|
||||
}
|
||||
|
||||
function promptHistorySteerId(item: PromptHistoryItem): string | null {
|
||||
const record = item as PromptHistoryItem & { steerId?: unknown };
|
||||
return normalizeSteerId(record.steerId);
|
||||
}
|
||||
|
||||
function promptHistoryMatch(item: PromptHistoryItem, steerId: string): SteerTraceMatch | null {
|
||||
if (promptHistorySteerId(item) !== steerId) return null;
|
||||
const prompt = normalizeSteerPromptText(String(item.text || ""));
|
||||
return {
|
||||
seq: item.seq,
|
||||
at: item.at,
|
||||
method: "turn/steer",
|
||||
steerId,
|
||||
promptChars: prompt.length,
|
||||
promptHash: steerPromptHash(prompt),
|
||||
promptOmitted: true,
|
||||
source: "promptHistory",
|
||||
};
|
||||
}
|
||||
|
||||
function outputMatch(item: LiveOutput, steerId: string): SteerTraceMatch | null {
|
||||
if (item.channel !== "user" || item.method !== "turn/steer" || normalizeSteerId(item.itemId) !== steerId) return null;
|
||||
const prompt = normalizeSteerPromptText(item.text);
|
||||
return {
|
||||
seq: item.seq,
|
||||
at: item.at,
|
||||
method: "turn/steer",
|
||||
steerId,
|
||||
promptChars: prompt.length,
|
||||
promptHash: steerPromptHash(prompt),
|
||||
promptOmitted: true,
|
||||
source: "output",
|
||||
};
|
||||
}
|
||||
|
||||
export function findSteerTraceConfirmation(task: QueueTask, steerId: string, output: LiveOutput[] = task.output): SteerTraceConfirmation {
|
||||
const matches = [
|
||||
...(Array.isArray(task.promptHistory) ? task.promptHistory : []).map((item) => promptHistoryMatch(item, steerId)),
|
||||
...output.map((item) => outputMatch(item, steerId)),
|
||||
]
|
||||
.filter((item): item is SteerTraceMatch => item !== null)
|
||||
.sort((left, right) => left.seq - right.seq);
|
||||
const bySeq = new Map<number, SteerTraceMatch>();
|
||||
for (const match of matches) {
|
||||
const existing = bySeq.get(match.seq);
|
||||
if (existing === undefined || existing.source === "promptHistory") bySeq.set(match.seq, match);
|
||||
}
|
||||
const uniqueMatches = Array.from(bySeq.values()).sort((left, right) => left.seq - right.seq);
|
||||
const trace = uniqueMatches[0] ?? null;
|
||||
return {
|
||||
taskId: task.id,
|
||||
steerId,
|
||||
found: trace !== null,
|
||||
accepted: trace !== null,
|
||||
deliveryState: trace !== null ? "accepted" : "unknown",
|
||||
trace,
|
||||
matches: uniqueMatches,
|
||||
duplicateSuppressionKey: steerId,
|
||||
};
|
||||
}
|
||||
|
||||
export function steerDuplicateDecision(task: QueueTask, steerId: string, prompt: string, output: LiveOutput[] = task.output): SteerDuplicateDecision {
|
||||
const confirmation = findSteerTraceConfirmation(task, steerId, output);
|
||||
const requestedPromptHash = steerPromptHash(prompt);
|
||||
const existingPromptHash = confirmation.trace?.promptHash ?? null;
|
||||
return {
|
||||
duplicate: confirmation.found && existingPromptHash === requestedPromptHash,
|
||||
conflict: confirmation.found && existingPromptHash !== requestedPromptHash,
|
||||
confirmation,
|
||||
existingPromptHash,
|
||||
requestedPromptHash,
|
||||
};
|
||||
}
|
||||
|
||||
export function steerTraceConfirmationJson(confirmation: SteerTraceConfirmation): JsonValue {
|
||||
return {
|
||||
taskId: confirmation.taskId,
|
||||
steerId: confirmation.steerId,
|
||||
found: confirmation.found,
|
||||
accepted: confirmation.accepted,
|
||||
deliveryState: confirmation.deliveryState,
|
||||
trace: confirmation.trace as unknown as JsonValue,
|
||||
matchCount: confirmation.matches.length,
|
||||
matches: confirmation.matches.slice(0, 5) as unknown as JsonValue,
|
||||
matchesTruncated: confirmation.matches.length > 5,
|
||||
duplicateSuppressionKey: confirmation.duplicateSuppressionKey,
|
||||
promptOmitted: true,
|
||||
};
|
||||
}
|
||||
@@ -380,6 +380,7 @@ export interface PromptHistoryItem {
|
||||
at: string;
|
||||
method: "turn/steer";
|
||||
text: string;
|
||||
steerId?: string;
|
||||
}
|
||||
|
||||
export interface QueueTask {
|
||||
|
||||
Reference in New Issue
Block a user