Pi runner: avoid global lock for compaction

Co-authored-by: kuna <47035140+1kuna@users.noreply.github.com>

src/agents/pi-embedded-runner/compact.ts

@@ -100,6 +100,7 @@ export type CompactEmbeddedPiSessionParams = {
   enqueue?: typeof enqueueCommand;
   extraSystemPrompt?: string;
   ownerNumbers?: string[];
+  skipSkillEnvOverrides?: boolean;
 };
 
 /**
@@ -110,7 +111,6 @@ export async function compactEmbeddedPiSessionDirect(
   params: CompactEmbeddedPiSessionParams,
 ): Promise<EmbeddedPiCompactResult> {
   const resolvedWorkspace = resolveUserPath(params.workspaceDir);
-  const prevCwd = process.cwd();
 
   const provider = (params.provider ?? DEFAULT_PROVIDER).trim() || DEFAULT_PROVIDER;
   const modelId = (params.model ?? DEFAULT_MODEL).trim() || DEFAULT_MODEL;
@@ -179,13 +179,14 @@ export async function compactEmbeddedPiSessionDirect(
     cwd: effectiveWorkspace,
   });
 
+  const shouldApplySkillEnvOverrides = !params.skipSkillEnvOverrides;
   let restoreSkillEnv: (() => void) | undefined;
-  process.chdir(effectiveWorkspace);
   try {
     const shouldLoadSkillEntries = !params.skillsSnapshot || !params.skillsSnapshot.resolvedSkills;
     const skillEntries = shouldLoadSkillEntries
       ? loadWorkspaceSkillEntries(effectiveWorkspace)
       : [];
+    if (shouldApplySkillEnvOverrides) {
       restoreSkillEnv = params.skillsSnapshot
         ? applySkillEnvOverridesFromSnapshot({
             snapshot: params.skillsSnapshot,
@@ -195,6 +196,7 @@ export async function compactEmbeddedPiSessionDirect(
             skills: skillEntries ?? [],
             config: params.config,
           });
+    }
     const skillsPrompt = resolveSkillsPromptForRun({
       skillsSnapshot: params.skillsSnapshot,
       entries: shouldLoadSkillEntries ? skillEntries : undefined,
@@ -318,7 +320,7 @@ export async function compactEmbeddedPiSessionDirect(
     const docsPath = await resolveMoltbotDocsPath({
       workspaceDir: effectiveWorkspace,
       argv1: process.argv[1],
-      cwd: process.cwd(),
+      cwd: effectiveWorkspace,
       moduleUrl: import.meta.url,
     });
     const ttsHint = params.config ? buildTtsSystemPromptHint(params.config) : undefined;
@@ -466,7 +468,6 @@ export async function compactEmbeddedPiSessionDirect(
     };
   } finally {
     restoreSkillEnv?.();
-    process.chdir(prevCwd);
   }
 }
 

src/agents/pi-embedded-runner/run.overflow-compaction.test.ts

@@ -168,6 +168,7 @@ function makeAttemptResult(
     didSendViaMessagingTool: false,
     messagingToolSentTexts: [],
     messagingToolSentTargets: [],
+    autoCompactionAttempts: 0,
     cloudCodeAssistFormatError: false,
     ...overrides,
   };
@@ -220,6 +221,20 @@ describe("overflow compaction in run loop", () => {
     expect(result.meta.error).toBeUndefined();
   });
 
+  it("skips manual compaction when auto-compaction already ran", async () => {
+    const overflowError = new Error("request_too_large: Request size exceeds model context window");
+
+    mockedRunEmbeddedAttempt.mockResolvedValueOnce(
+      makeAttemptResult({ promptError: overflowError, autoCompactionAttempts: 1 }),
+    );
+
+    const result = await runEmbeddedPiAgent(baseParams);
+
+    expect(mockedCompactDirect).not.toHaveBeenCalled();
+    expect(mockedRunEmbeddedAttempt).toHaveBeenCalledTimes(1);
+    expect(result.meta.error?.kind).toBe("context_overflow");
+  });
+
   it("returns error if compaction fails", async () => {
     const overflowError = new Error("request_too_large: Request size exceeds model context window");
 

src/agents/pi-embedded-runner/run.ts

@@ -86,8 +86,6 @@ export async function runEmbeddedPiAgent(
       : "markdown");
   const isProbeSession = params.sessionId?.startsWith("probe-") ?? false;
 
-  return enqueueSession(() =>
-    enqueueGlobal(async () => {
   const started = Date.now();
   const resolvedWorkspace = resolveUserPath(params.workspaceDir);
   const prevCwd = process.cwd();
@@ -95,8 +93,7 @@ export async function runEmbeddedPiAgent(
   const provider = (params.provider ?? DEFAULT_PROVIDER).trim() || DEFAULT_PROVIDER;
   const modelId = (params.model ?? DEFAULT_MODEL).trim() || DEFAULT_MODEL;
   const agentDir = params.agentDir ?? resolveMoltbotAgentDir();
-      const fallbackConfigured =
+  const fallbackConfigured = (params.config?.agents?.defaults?.model?.fallbacks?.length ?? 0) > 0;
-        (params.config?.agents?.defaults?.model?.fallbacks?.length ?? 0) > 0;
   await ensureMoltbotModelsJson(params.config, agentDir);
 
   const { model, error, authStorage, modelRegistry } = resolveModel(
@@ -229,8 +226,7 @@ export async function runEmbeddedPiAgent(
       return;
     }
     if (model.provider === "github-copilot") {
-          const { resolveCopilotApiToken } =
+      const { resolveCopilotApiToken } = await import("../../providers/github-copilot-token.js");
-            await import("../../providers/github-copilot-token.js");
       const copilotToken = await resolveCopilotApiToken({
         githubToken: apiKeyInfo.apiKey,
       });
@@ -267,11 +263,7 @@ export async function runEmbeddedPiAgent(
   try {
     while (profileIndex < profileCandidates.length) {
       const candidate = profileCandidates[profileIndex];
-          if (
+      if (candidate && candidate !== lockedProfileId && isProfileInCooldown(authStore, candidate)) {
-            candidate &&
-            candidate !== lockedProfileId &&
-            isProfileInCooldown(authStore, candidate)
-          ) {
         profileIndex += 1;
         continue;
       }
@@ -301,7 +293,9 @@ export async function runEmbeddedPiAgent(
       const prompt =
         provider === "anthropic" ? scrubAnthropicRefusalMagic(params.prompt) : params.prompt;
 
-          const attempt = await runEmbeddedAttempt({
+      const attempt = await enqueueSession(() =>
+        enqueueGlobal(async () =>
+          runEmbeddedAttempt({
             sessionId: params.sessionId,
             sessionKey: params.sessionKey,
             messageChannel: params.messageChannel,
@@ -354,21 +348,35 @@ export async function runEmbeddedPiAgent(
             streamParams: params.streamParams,
             ownerNumbers: params.ownerNumbers,
             enforceFinalTag: params.enforceFinalTag,
-          });
+          }),
+        ),
+      );
 
-          const { aborted, promptError, timedOut, sessionIdUsed, lastAssistant } = attempt;
+      const {
+        aborted,
+        promptError,
+        timedOut,
+        sessionIdUsed,
+        lastAssistant,
+        autoCompactionAttempts,
+      } = attempt;
 
       if (promptError && !aborted) {
         const errorText = describeUnknownError(promptError);
         if (isContextOverflowError(errorText)) {
           const isCompactionFailure = isCompactionFailureError(errorText);
           // Attempt auto-compaction on context overflow (not compaction_failure)
-              if (!isCompactionFailure && !overflowCompactionAttempted) {
+          if (
+            !isCompactionFailure &&
+            !overflowCompactionAttempted &&
+            autoCompactionAttempts === 0
+          ) {
             log.warn(
               `context overflow detected; attempting auto-compaction for ${provider}/${modelId}`,
             );
             overflowCompactionAttempted = true;
-                const compactResult = await compactEmbeddedPiSessionDirect({
+            const compactResult = await enqueueSession(() =>
+              compactEmbeddedPiSessionDirect({
                 sessionId: params.sessionId,
                 sessionKey: params.sessionKey,
                 messageChannel: params.messageChannel,
@@ -387,7 +395,9 @@ export async function runEmbeddedPiAgent(
                 bashElevated: params.bashElevated,
                 extraSystemPrompt: params.extraSystemPrompt,
                 ownerNumbers: params.ownerNumbers,
-                });
+                skipSkillEnvOverrides: true,
+              }),
+            );
             if (compactResult.compacted) {
               log.info(`auto-compaction succeeded for ${provider}/${modelId}; retrying prompt`);
               continue;
@@ -674,6 +684,4 @@ export async function runEmbeddedPiAgent(
   } finally {
     process.chdir(prevCwd);
   }
-    }),
-  );
 }

src/agents/pi-embedded-runner/run/attempt.ts

@@ -619,6 +619,7 @@ export async function runEmbeddedAttempt(
         toolMetas,
         unsubscribe,
         waitForCompactionRetry,
+        getCompactionAttempts,
         getMessagingToolSentTexts,
         getMessagingToolSentTargets,
         didSendViaMessagingTool,
@@ -865,6 +866,7 @@ export async function runEmbeddedAttempt(
         didSendViaMessagingTool: didSendViaMessagingTool(),
         messagingToolSentTexts: getMessagingToolSentTexts(),
         messagingToolSentTargets: getMessagingToolSentTargets(),
+        autoCompactionAttempts: getCompactionAttempts(),
         cloudCodeAssistFormatError: Boolean(
           lastAssistant?.errorMessage && isCloudCodeAssistFormatError(lastAssistant.errorMessage),
         ),

src/agents/pi-embedded-runner/run/types.ts

@@ -102,6 +102,7 @@ export type EmbeddedRunAttemptResult = {
   didSendViaMessagingTool: boolean;
   messagingToolSentTexts: string[];
   messagingToolSentTargets: MessagingToolSend[];
+  autoCompactionAttempts: number;
   cloudCodeAssistFormatError: boolean;
   /** Client tool call detected (OpenResponses hosted tools). */
   clientToolCall?: { name: string; params: Record<string, unknown> };

src/agents/pi-embedded-subscribe.handlers.lifecycle.ts

@@ -22,6 +22,7 @@ export function handleAgentStart(ctx: EmbeddedPiSubscribeContext) {
 
 export function handleAutoCompactionStart(ctx: EmbeddedPiSubscribeContext) {
   ctx.state.compactionInFlight = true;
+  ctx.state.compactionAttempts += 1;
   ctx.ensureCompactionPromise();
   ctx.log.debug(`embedded run compaction start: runId=${ctx.params.runId}`);
   emitAgentEvent({

src/agents/pi-embedded-subscribe.handlers.types.ts

@@ -49,6 +49,7 @@ export type EmbeddedPiSubscribeState = {
   lastReasoningSent?: string;
 
   compactionInFlight: boolean;
+  compactionAttempts: number;
   pendingCompactionRetry: number;
   compactionRetryResolve?: () => void;
   compactionRetryPromise: Promise<void> | null;

src/agents/pi-embedded-subscribe.ts

@@ -57,6 +57,7 @@ export function subscribeEmbeddedPiSession(params: SubscribeEmbeddedPiSessionPar
     suppressBlockChunks: false, // Avoid late chunk inserts after final text merge.
     lastReasoningSent: undefined,
     compactionInFlight: false,
+    compactionAttempts: 0,
     pendingCompactionRetry: 0,
     compactionRetryResolve: undefined,
     compactionRetryPromise: null,
@@ -472,6 +473,7 @@ export function subscribeEmbeddedPiSession(params: SubscribeEmbeddedPiSessionPar
     toolMetas,
     unsubscribe,
     isCompacting: () => state.compactionInFlight || state.pendingCompactionRetry > 0,
+    getCompactionAttempts: () => state.compactionAttempts,
     getMessagingToolSentTexts: () => messagingToolSentTexts.slice(),
     getMessagingToolSentTargets: () => messagingToolSentTargets.slice(),
     // Returns true if any messaging tool successfully sent a message.