Add Code Queue resume contract

2026-05-23 10:28:57 +00:00
parent 184667b4cb
commit fd8e21e611
11 changed files with 830 additions and 4 deletions
@@ -402,7 +402,7 @@ function codeQueueK3sServiceIdForRequest(method: string, targetPath: string): st
  if (targetPath === "/api/queues" || targetPath === "/api/tasks/overview") return "code-queue-scheduler";
  if (targetPath === "/api/oa/backfill" || targetPath === "/api/notifications/claudeqq/drain" || targetPath === "/api/notifications/claudeqq/backfill") return "code-queue-scheduler";
  if (targetPath === "/api/judge/probe" || targetPath === "/api/judge/self-test" || targetPath === "/api/queue-order/self-test" || targetPath === "/api/reference-injection/self-test" || targetPath === "/api/trace-port/self-test") return "code-queue-scheduler";
-  if (/^\/api\/tasks\/[^/]+\/(?:steer|interrupt)$/u.test(targetPath)) return "code-queue-scheduler";
+  if (/^\/api\/tasks\/[^/]+\/(?:steer|resume|interrupt)$/u.test(targetPath)) return "code-queue-scheduler";
  if (/^\/api\/tasks\/[^/]+$/u.test(targetPath) && normalizedMethod === "DELETE") return "code-queue-scheduler";
  if (targetPath === "/api/dev-containers" || /^\/api\/dev-containers(?:\/[^/]+)?\/start$/u.test(targetPath)) return "code-queue-scheduler";
  if (/^\/api\/dev-containers(?:\/[^/]+)?\/status$/u.test(targetPath)) return "code-queue-scheduler";
@@ -137,6 +137,7 @@ import {
 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 { createResumeId, findResumeTraceConfirmation, normalizeResumeId, resumeDuplicateDecision, resumePromptHash, resumeTraceConfirmationJson, resumeTraceText } from "./resume-confirmation";
 import { configureSelfTests, runJudgeInfraSelfTest, runQueueClaimMoveSelfTest, runQueueOrderingSelfTest, runReferenceInjectionSelfTest, runStaleActiveRecoverySelfTest, runTracePortSelfTest, runTraceSummaryContractSelfTest } from "./self-tests";
 import {
  codexToolLifecycleStartedBeforeIn,
@@ -4873,6 +4874,107 @@ async function steerTask(task: QueueTask, req: Request): Promise<Response> {
  });
 }

+async function resumeTask(task: QueueTask, req: Request): Promise<Response> {
+  if (!serviceRoleAllowsScheduler(config.serviceRole)) return schedulerOnlyRejectResponse(req.method, `/api/tasks/${task.id}/resume`);
+  const notReady = requireDatabaseReadyForWrite(req.method, `/api/tasks/${task.id}/resume`);
+  if (notReady !== null) return notReady;
+  const body = await readJson(req);
+  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 resumeId = normalizeResumeId(bodyRecord.resumeId) ?? createResumeId(task.id, prompt);
+  const originalCodexThreadId = task.codexThreadId;
+  const duplicateDecision = resumeDuplicateDecision(task, resumeId, prompt, taskFullOutput(task));
+  if (duplicateDecision.duplicate) {
+    await flushDirtyTasksToDatabase(true);
+    return jsonResponse({
+      ok: true,
+      accepted: true,
+      duplicateSuppressed: true,
+      deliveryState: "duplicate_suppressed",
+      resumeId,
+      reuseOriginalThread: originalCodexThreadId !== null,
+      originalCodexThreadId,
+      codexThreadId: task.codexThreadId,
+      traceConfirmation: resumeTraceConfirmationJson(duplicateDecision.confirmation),
+      task: taskForResponse(task),
+      queue: await queueSummaryForResponse(),
+    });
+  }
+  if (duplicateDecision.conflict) {
+    return jsonResponse({
+      ok: false,
+      error: "resumeId already exists with a different prompt hash",
+      accepted: false,
+      deliveryState: "not_accepted",
+      resumeId,
+      existingPromptHash: duplicateDecision.existingPromptHash,
+      requestedPromptHash: duplicateDecision.requestedPromptHash,
+      traceConfirmation: resumeTraceConfirmationJson(duplicateDecision.confirmation),
+      task: taskForResponse(task),
+    }, 409);
+  }
+  if (task.status === "running" || task.status === "judging") {
+    return jsonResponse({
+      ok: false,
+      error: `task is active: ${task.status}`,
+      accepted: false,
+      deliveryState: "not_accepted",
+      disposition: "use-steer-for-active-task",
+      resumeId,
+      task: taskForResponse(task),
+      commands: {
+        steer: `bun scripts/cli.ts codex steer ${task.id} --prompt-file <path>`,
+        show: `bun scripts/cli.ts codex task ${task.id}`,
+      },
+    }, 409);
+  }
+  if (task.status !== "failed" && task.status !== "canceled" && task.status !== "succeeded") {
+    return jsonResponse({
+      ok: false,
+      error: `task is not resumable: ${task.status}`,
+      accepted: false,
+      deliveryState: "not_accepted",
+      disposition: "not-terminal",
+      resumeId,
+      task: taskForResponse(task),
+    }, 409);
+  }
+  const traceOutput = appendOutput(task, "user", resumeTraceText(resumeId, prompt), "turn/resume", resumeId);
+  task.status = "queued";
+  task.finishedAt = null;
+  task.readAt = null;
+  task.cancelRequested = false;
+  task.lastError = null;
+  task.maxAttempts = Math.max(task.maxAttempts, task.attempts.length + 1);
+  task.nextMode = "retry";
+  task.nextPrompt = prompt;
+  setAttemptFeedbackPrompt(task.attempts.at(-1), task.nextPrompt, "manual-resume", task.attempts.length + 1);
+  task.updatedAt = nowIso();
+  task.queueEnteredAt = task.updatedAt;
+  appendOutput(task, "system", `resume queued resumeId=${resumeId} reuseOriginalThread=${originalCodexThreadId !== null ? "true" : "false"}\n`, "manual-resume", resumeId);
+  armIdleNotification();
+  persistState();
+  scheduleQueue(queueIdOf(task));
+  await flushDirtyTasksToDatabase(true);
+  const traceConfirmation = findResumeTraceConfirmation(task, resumeId, taskFullOutput(task));
+  return jsonResponse({
+    ok: true,
+    accepted: true,
+    duplicateSuppressed: false,
+    deliveryState: traceConfirmation.found ? "queued_for_existing_thread" : "accepted_response_timeout",
+    resumeId,
+    turnId: traceOutput?.seq ?? null,
+    promptHash: resumePromptHash(prompt),
+    reuseOriginalThread: originalCodexThreadId !== null,
+    originalCodexThreadId,
+    codexThreadId: task.codexThreadId,
+    traceConfirmation: resumeTraceConfirmationJson(traceConfirmation),
+    task: taskForResponse(task),
+    queue: await queueSummaryForResponse(),
+  }, 202);
+}
+
 async function editQueuedTaskPrompt(task: QueueTask, req: Request): Promise<Response> {
  if (!serviceRoleAllowsWrite(config.serviceRole)) return readOnlyRejectResponse(req.method, `/api/tasks/${task.id}/edit`);
  const notReady = requireDatabaseReadyForWrite(req.method, `/api/tasks/${task.id}/edit`);
@@ -5797,7 +5899,7 @@ async function route(req: Request): Promise<Response> {
      if (task === null) return jsonResponse({ ok: false, error: "task not found" }, 404);
      return jsonResponse({ ok: true, summary: taskSummaryResponse(task, url) });
    }
-    const match = url.pathname.match(/^\/api\/tasks\/([^/]+)(?:\/(retry|steer|interrupt|move|read|edit))?$/u);
+    const match = url.pathname.match(/^\/api\/tasks\/([^/]+)(?:\/(retry|resume|steer|interrupt|move|read|edit))?$/u);
    if (match !== null) {
      const action = match[2];
      const taskId = decodeURIComponent(match[1] ?? "");
@@ -5812,6 +5914,7 @@ async function route(req: Request): Promise<Response> {
        : await findTaskForMutation(taskId);
      if (task === null) return jsonResponse({ ok: false, error: "task not found" }, 404);
      if (action === "retry" && req.method === "POST") return await manualRetry(task, req);
+      if (action === "resume" && req.method === "POST") return await resumeTask(task, req);
      if (action === "steer" && req.method === "POST") return await steerTask(task, req);
      if (action === "interrupt" && req.method === "POST") return await interruptTask(task, req.method);
      if (action === "move" && req.method === "POST") return await moveTaskToQueue(task, req);
@@ -0,0 +1,142 @@
+import { createHash } from "node:crypto";
+import type { JsonValue, LiveOutput, QueueTask } from "./types";
+
+const resumeIdPattern = /^[A-Za-z0-9._:-]{8,128}$/u;
+
+export type ResumeDeliveryState = "queued_for_existing_thread" | "duplicate_suppressed" | "not_accepted" | "accepted_response_timeout" | "unknown";
+
+export interface ResumeTraceMatch {
+  seq: number;
+  at: string;
+  method: "turn/resume";
+  resumeId: string;
+  promptChars: number;
+  promptHash: string;
+  promptOmitted: true;
+  source: "output";
+}
+
+export interface ResumeTraceConfirmation {
+  taskId: string;
+  resumeId: string;
+  found: boolean;
+  accepted: boolean;
+  deliveryState: ResumeDeliveryState;
+  trace: ResumeTraceMatch | null;
+  matches: ResumeTraceMatch[];
+  duplicateSuppressionKey: string;
+}
+
+export interface ResumeDuplicateDecision {
+  duplicate: boolean;
+  conflict: boolean;
+  confirmation: ResumeTraceConfirmation;
+  existingPromptHash: string | null;
+  requestedPromptHash: string;
+}
+
+export function normalizeResumeId(value: unknown): string | null {
+  if (typeof value !== "string") return null;
+  const text = value.trim();
+  if (!resumeIdPattern.test(text)) return null;
+  return text;
+}
+
+export function resumePromptHash(prompt: string): string {
+  return createHash("sha256").update(prompt, "utf8").digest("hex");
+}
+
+export function createResumeId(taskId: string, prompt: string): string {
+  const hash = createHash("sha256")
+    .update("unidesk-code-queue-resume:v1", "utf8")
+    .update("\0", "utf8")
+    .update(taskId, "utf8")
+    .update("\0", "utf8")
+    .update(prompt, "utf8")
+    .digest("hex")
+    .slice(0, 24);
+  return `resume_${hash}`;
+}
+
+export function resumeTraceText(resumeId: string, prompt: string): string {
+  return `\n[resume id=${resumeId} chars=${prompt.length} sha256=${resumePromptHash(prompt)}] ${prompt}\n`;
+}
+
+export function normalizeResumePromptText(text: string): string {
+  return text.replace(/^\s*\[resume(?:\s+[^\]]*)?\]\s*/u, "").trimEnd();
+}
+
+function resumeTraceMeta(text: string): { promptChars: number | null; promptHash: string | null } {
+  const match = text.match(/^\s*\[resume(?:\s+[^\]]*)?\]\s*/u);
+  if (match === null) return { promptChars: null, promptHash: null };
+  const header = match[0];
+  const charsMatch = header.match(/\schars=(\d+)(?=\s|\])/u);
+  const hashMatch = header.match(/\ssha256=([a-f0-9]{64})(?=\s|\])/u);
+  return {
+    promptChars: charsMatch === null ? null : Number(charsMatch[1]),
+    promptHash: hashMatch?.[1] ?? null,
+  };
+}
+
+function outputMatch(item: LiveOutput, resumeId: string): ResumeTraceMatch | null {
+  if (item.channel !== "user" || item.method !== "turn/resume" || normalizeResumeId(item.itemId) !== resumeId) return null;
+  const prompt = normalizeResumePromptText(item.text);
+  const meta = resumeTraceMeta(item.text);
+  return {
+    seq: item.seq,
+    at: item.at,
+    method: "turn/resume",
+    resumeId,
+    promptChars: meta.promptChars ?? prompt.length,
+    promptHash: meta.promptHash ?? resumePromptHash(prompt),
+    promptOmitted: true,
+    source: "output",
+  };
+}
+
+export function findResumeTraceConfirmation(task: QueueTask, resumeId: string, output: LiveOutput[] = task.output): ResumeTraceConfirmation {
+  const matches = output
+    .map((item) => outputMatch(item, resumeId))
+    .filter((item): item is ResumeTraceMatch => item !== null)
+    .sort((left, right) => left.seq - right.seq);
+  const trace = matches[0] ?? null;
+  return {
+    taskId: task.id,
+    resumeId,
+    found: trace !== null,
+    accepted: trace !== null,
+    deliveryState: trace !== null ? "queued_for_existing_thread" : "unknown",
+    trace,
+    matches,
+    duplicateSuppressionKey: resumeId,
+  };
+}
+
+export function resumeDuplicateDecision(task: QueueTask, resumeId: string, prompt: string, output: LiveOutput[] = task.output): ResumeDuplicateDecision {
+  const confirmation = findResumeTraceConfirmation(task, resumeId, output);
+  const requestedPromptHash = resumePromptHash(prompt);
+  const existingPromptHash = confirmation.trace?.promptHash ?? null;
+  return {
+    duplicate: confirmation.found && existingPromptHash === requestedPromptHash,
+    conflict: confirmation.found && existingPromptHash !== requestedPromptHash,
+    confirmation,
+    existingPromptHash,
+    requestedPromptHash,
+  };
+}
+
+export function resumeTraceConfirmationJson(confirmation: ResumeTraceConfirmation): JsonValue {
+  return {
+    taskId: confirmation.taskId,
+    resumeId: confirmation.resumeId,
+    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,
+  };
+}
@@ -1157,6 +1157,7 @@ function buildTaskTranscript(task: QueueTask, limit = 180, rawOutputWindow = 0,
  for (const item of outputItems) {
    if (initialPrompt !== null && item.channel === "user" && item.method === "enqueue") continue;
    if (item.channel === "user" && item.method === "turn/steer" && promptHistorySeqs.has(item.seq)) continue;
+    if (item.channel === "user" && item.method === "turn/resume") continue;
    if (isOpenCodeStepBoundaryMethod(item.method)) continue;
    if (item.channel === "command" && item.method === "item/started") {
      flushMessage();