fix(agents): prioritize manual session turns

galiniliev · Galin Iliev · commit 2bc5e9ed5a25 · 2026-05-17T00:40:13.000Z
diff --git a/src/agents/pi-embedded-runner/run.overflow-compaction.test.ts b/src/agents/pi-embedded-runner/run.overflow-compaction.test.ts
@@ -307,6 +307,40 @@ describe("runEmbeddedPiAgent overflow compaction trigger routing", () => {
     expect(attemptParams?.internalEvents).toBe(internalEvents);
   });
 
+  it("marks user-triggered session queue work as foreground", async () => {
+    mockedRunEmbeddedAttempt.mockResolvedValueOnce(makeAttemptResult({ promptError: null }));
+    const observedPriorities: unknown[] = [];
+
+    await runEmbeddedPiAgent({
+      ...overflowBaseRunParams,
+      trigger: "user",
+      runId: "run-user-session-priority",
+      enqueue: async (task, opts) => {
+        observedPriorities.push(opts?.priority);
+        return await task();
+      },
+    });
+
+    expect(observedPriorities[0]).toBe("foreground");
+  });
+
+  it("marks cron-triggered session queue work as background", async () => {
+    mockedRunEmbeddedAttempt.mockResolvedValueOnce(makeAttemptResult({ promptError: null }));
+    const observedPriorities: unknown[] = [];
+
+    await runEmbeddedPiAgent({
+      ...overflowBaseRunParams,
+      trigger: "cron",
+      runId: "run-cron-session-priority",
+      enqueue: async (task, opts) => {
+        observedPriorities.push(opts?.priority);
+        return await task();
+      },
+    });
+
+    expect(observedPriorities[0]).toBe("background");
+  });
+
   it("forwards explicit OpenAI Codex auth profiles to codex plugin harnesses", async () => {
     const { clearAgentHarnesses, registerAgentHarness } = await import("../harness/registry.js");
     const pluginRunAttempt = vi.fn<AgentHarness["runAttempt"]>(async () =>
diff --git a/src/agents/pi-embedded-runner/run.ts b/src/agents/pi-embedded-runner/run.ts
@@ -211,6 +211,23 @@ function withEmbeddedRunLaneTimeout(
   return { ...opts, taskTimeoutMs: laneTaskTimeoutMs };
 }
 
+function resolveEmbeddedRunSessionQueuePriority(
+  trigger: RunEmbeddedPiAgentParams["trigger"],
+): CommandQueueEnqueueOptions["priority"] {
+  switch (trigger) {
+    case "user":
+    case "manual":
+      return "foreground";
+    case "cron":
+    case "heartbeat":
+    case "memory":
+    case "overflow":
+      return "background";
+    default:
+      return "normal";
+  }
+}
+
 function normalizeEmbeddedRunAttemptResult(
   attempt: EmbeddedRunAttemptForRunner,
 ): EmbeddedRunAttemptForRunner {
@@ -378,6 +395,7 @@ export async function runEmbeddedPiAgent(
   }
   const sessionLane = resolveSessionLane(params.sessionKey?.trim() || params.sessionId);
   const globalLane = resolveGlobalLane(params.lane);
+  const sessionQueuePriority = resolveEmbeddedRunSessionQueuePriority(params.trigger);
   const laneTaskTimeoutMs = resolveEmbeddedRunLaneTimeoutMs(params.timeoutMs);
   let laneTaskProgressAtMs = Date.now();
   const noteLaneTaskProgress = () => {
@@ -395,8 +413,12 @@ export async function runEmbeddedPiAgent(
     params.enqueue
       ? params.enqueue(task, withLaneTimeout(opts))
       : enqueueCommandInLane(globalLane, task, withLaneTimeout(opts));
-  const enqueueSession = <T>(task: () => Promise<T>, opts?: CommandQueueEnqueueOptions) =>
-    params.enqueue ? params.enqueue(task, opts) : enqueueCommandInLane(sessionLane, task, opts);
+  const enqueueSession = <T>(task: () => Promise<T>, opts?: CommandQueueEnqueueOptions) => {
+    const sessionOpts: CommandQueueEnqueueOptions = { ...opts, priority: sessionQueuePriority };
+    return params.enqueue
+      ? params.enqueue(task, sessionOpts)
+      : enqueueCommandInLane(sessionLane, task, sessionOpts);
+  };
   const channelHint = params.messageChannel ?? params.messageProvider;
   const resolvedToolResultFormat =
     params.toolResultFormat ??
diff --git a/src/process/command-queue.test.ts b/src/process/command-queue.test.ts
@@ -161,6 +161,64 @@ describe("command queue", () => {
     expect(getQueueSize()).toBe(0);
   });
 
+  it("runs foreground work before already queued background work", async () => {
+    const { task: blocker, release } = enqueueBlockedMainTask(async () => "blocker");
+    const calls: string[] = [];
+
+    const background = enqueueCommandInLane(
+      CommandLane.Main,
+      async () => {
+        calls.push("background");
+        return "background";
+      },
+      { priority: "background" },
+    );
+    const normal = enqueueCommandInLane(CommandLane.Main, async () => {
+      calls.push("normal");
+      return "normal";
+    });
+    const foreground = enqueueCommandInLane(
+      CommandLane.Main,
+      async () => {
+        calls.push("foreground");
+        return "foreground";
+      },
+      { priority: "foreground" },
+    );
+
+    release();
+    await expect(blocker).resolves.toBe("blocker");
+    await expect(foreground).resolves.toBe("foreground");
+    await expect(normal).resolves.toBe("normal");
+    await expect(background).resolves.toBe("background");
+    expect(calls).toEqual(["foreground", "normal", "background"]);
+  });
+
+  it("preserves FIFO order within each priority", async () => {
+    const { task: blocker, release } = enqueueBlockedMainTask(async () => "blocker");
+    const calls: string[] = [];
+
+    const first = enqueueCommandInLane(
+      CommandLane.Main,
+      async () => {
+        calls.push("first");
+      },
+      { priority: "foreground" },
+    );
+    const second = enqueueCommandInLane(
+      CommandLane.Main,
+      async () => {
+        calls.push("second");
+      },
+      { priority: "foreground" },
+    );
+
+    release();
+    await blocker;
+    await Promise.all([first, second]);
+    expect(calls).toEqual(["first", "second"]);
+  });
+
   it("logs enqueue depth after push", async () => {
     const task = enqueueCommand(async () => {});
 
diff --git a/src/process/command-queue.ts b/src/process/command-queue.ts
@@ -61,6 +61,8 @@ type QueueEntry = {
   resolve: (value: unknown) => void;
   reject: (reason?: unknown) => void;
   enqueuedAt: number;
+  sequence: number;
+  priority: number;
   warnAfterMs: number;
   taskTimeoutMs?: number;
   taskTimeoutProgressAtMs?: () => number | undefined;
@@ -107,6 +109,7 @@ function getQueueState() {
     lanes: new Map<string, LaneState>(),
     activeTaskWaiters: new Set<ActiveTaskWaiter>(),
     nextTaskId: 1,
+    nextQueueSequence: 1,
   }));
   // Schema migration: the singleton may have been created by an older code
   // version (e.g. v2026.4.2) that did not include `activeTaskWaiters`.  After
@@ -117,6 +120,27 @@ function getQueueState() {
   if (!state.activeTaskWaiters) {
     state.activeTaskWaiters = new Set<ActiveTaskWaiter>();
   }
+  if (!state.nextQueueSequence) {
+    state.nextQueueSequence = 1;
+  }
+  let maxQueueSequence = state.nextQueueSequence - 1;
+  for (const lane of state.lanes.values()) {
+    for (const entry of lane.queue as Array<
+      QueueEntry & { priority?: number; sequence?: number }
+    >) {
+      if (typeof entry.priority !== "number") {
+        entry.priority = 0;
+      }
+      if (typeof entry.sequence !== "number") {
+        entry.sequence = state.nextQueueSequence++;
+      } else {
+        maxQueueSequence = Math.max(maxQueueSequence, entry.sequence);
+      }
+    }
+  }
+  if (state.nextQueueSequence <= maxQueueSequence) {
+    state.nextQueueSequence = maxQueueSequence + 1;
+  }
   return state;
 }
 
@@ -204,6 +228,30 @@ function normalizeTaskTimeoutMs(value: number | undefined): number | undefined {
   return Math.max(1, Math.floor(value));
 }
 
+function resolveQueuePriority(priority: CommandQueueEnqueueOptions["priority"]): number {
+  switch (priority) {
+    case "foreground":
+      return 1;
+    case "background":
+      return -1;
+    default:
+      return 0;
+  }
+}
+
+function enqueueLaneEntry(state: LaneState, entry: QueueEntry): void {
+  const insertAt = state.queue.findIndex(
+    (queued) =>
+      queued.priority < entry.priority ||
+      (queued.priority === entry.priority && queued.sequence > entry.sequence),
+  );
+  if (insertAt < 0) {
+    state.queue.push(entry);
+    return;
+  }
+  state.queue.splice(insertAt, 0, entry);
+}
+
 async function runQueueEntryTask(lane: string, entry: QueueEntry): Promise<unknown> {
   const taskPromise = Promise.resolve().then(entry.task);
   const taskTimeoutMs = normalizeTaskTimeoutMs(entry.taskTimeoutMs);
@@ -362,11 +410,13 @@ export function enqueueCommandInLane<T>(
   const warnAfterMs = opts?.warnAfterMs ?? 2_000;
   const state = getLaneState(cleaned);
   return new Promise<T>((resolve, reject) => {
-    state.queue.push({
+    enqueueLaneEntry(state, {
       task: () => task(),
       resolve: (value) => resolve(value as T),
       reject,
       enqueuedAt: Date.now(),
+      sequence: queueState.nextQueueSequence++,
+      priority: resolveQueuePriority(opts?.priority),
       warnAfterMs,
       taskTimeoutMs: normalizeTaskTimeoutMs(opts?.taskTimeoutMs),
       taskTimeoutProgressAtMs: opts?.taskTimeoutProgressAtMs,
@@ -472,6 +522,7 @@ export function resetCommandQueueStateForTest(): void {
     resolveActiveTaskWaiter(waiter, { drained: true });
   }
   queueState.nextTaskId = 1;
+  queueState.nextQueueSequence = 1;
 }
 
 /**
diff --git a/src/process/command-queue.types.ts b/src/process/command-queue.types.ts
@@ -3,6 +3,7 @@ export type CommandQueueEnqueueOptions = {
   onWait?: (waitMs: number, queuedAhead: number) => void;
   taskTimeoutMs?: number;
   taskTimeoutProgressAtMs?: () => number | undefined;
+  priority?: "foreground" | "normal" | "background";
 };
 
 export type CommandQueueEnqueueFn = <T>(
