feat: Enhance agent orchestration and implement session management

- Added logic to treat manual runs as approval handoffs in AgentOrchestrator.
- Refactored LLM timeout handling in ImplementSessionManager to allow configuration.
- Introduced methods to manage runner feedback and LLM request completion.
- Updated OpenAI responses clients to support threading for retries.
- Improved test coverage for agent orchestration and session management, including handling stale feedback and chaining retries.
- Adjusted responses client to include previous response IDs for better tracking.

Author: lhy0718

ISSUES.md

@@ -40,9 +40,26 @@ export class AgentOrchestrator {
     }
 
     if (current.currentNode !== nodeId) {
-      await this.runtime.jumpToNode(runId, nodeId, "force", "manual node run");
+      if (shouldTreatManualRunAsApprovalHandoff(current, nodeId)) {
+        await this.runtime.approveCurrent(runId, { continueAfterApprove: false });
+      } else {
+        await this.runtime.jumpToNode(runId, nodeId, "force", "manual node run");
+      }
     }
 
+
+function shouldTreatManualRunAsApprovalHandoff(run: RunRecord, nodeId: GraphNodeId): boolean {
+  const recommendation = run.graph.pendingTransition;
+  if (!recommendation || recommendation.action !== "pause_for_human") {
+    return false;
+  }
+  if (recommendation.targetNode !== nodeId) {
+    return false;
+  }
+  const currentIdx = GRAPH_NODE_ORDER.indexOf(run.currentNode);
+  const targetIdx = GRAPH_NODE_ORDER.indexOf(nodeId);
+  return currentIdx >= 0 && targetIdx === currentIdx + 1;
+}
     await this.runtime.runUntilPause(runId, {
       abortSignal: opts?.abortSignal,
       stopAfterApprovalBoundary: true,

src/core/agents/agentOrchestrator.ts

@@ -671,41 +671,22 @@ export class ImplementSessionManager {
             if (!this.deps.llm) {
               throw new Error("implement_experiments is configured for staged_llm mode, but no LLM client is available.");
             }
-            const llmTimeoutMs = getImplementLlmTimeoutMs();
-            const timeoutController = new AbortController();
-            const timeoutId = setTimeout(() => timeoutController.abort(), llmTimeoutMs);
-            const llmAbortSignal = abortSignal
-              ? AbortSignal.any([abortSignal, timeoutController.signal])
-              : timeoutController.signal;
-            try {
-              const completion = await this.deps.llm.complete(attemptPrompt, {
-                systemPrompt: attemptSystemPrompt,
-                abortSignal: llmAbortSignal,
-                onProgress: (event) => {
-                  const text = event.text.trim();
-                  if (!text) {
-                    return;
-                  }
-                  emitImplementObservation("codex", event.type === "delta" ? `LLM> ${text}` : text, {
-                    attempt,
-                    threadId: activeThreadId,
-                    publicDir: defaultPublicDir
-                  });
-                }
-              });
-              result = {
-                threadId: activeThreadId,
-                finalText: completion.text,
-                events: []
-              };
-            } catch (error) {
-              if (timeoutController.signal.aborted && !abortSignal?.aborted) {
-                throw new Error(`implement_experiments staged_llm request timed out after ${llmTimeoutMs}ms`);
-              }
-              throw error;
-            } finally {
-              clearTimeout(timeoutId);
-            }
+            const llmTimeoutMs = getImplementLlmTimeoutMs(this.deps.config);
+            const completion = await this.completeStagedLlmRequest({
+              prompt: attemptPrompt,
+              systemPrompt: attemptSystemPrompt,
+              timeoutMs: llmTimeoutMs,
+              abortSignal,
+              attempt,
+              threadId: activeThreadId,
+              publicDir: defaultPublicDir,
+              emitImplementObservation
+            });
+            result = {
+              threadId: completion.threadId || activeThreadId,
+              finalText: completion.text,
+              events: []
+            };
           }
         } catch (error) {
           const errorMessage = error instanceof Error ? error.message : String(error);
@@ -1368,9 +1349,7 @@ export class ImplementSessionManager {
     const previousSummary = await runContext.get<string>("implement_experiments.last_summary");
     const previousRunCommand = await runContext.get<string>("implement_experiments.run_command");
     const previousScript = await runContext.get<string>("implement_experiments.script");
-    const runnerFeedback =
-      (await runContext.get<RunVerifierReport>("implement_experiments.runner_feedback")) ||
-      (await runContext.get<RunVerifierReport>("run_experiments.feedback_for_implementer"));
+    const runnerFeedback = await this.loadApplicableRunnerFeedback(run, runContext);
     const paperCritique = await runContext.get<{
       overall_decision?: string;
       manuscript_type?: string;
@@ -1461,6 +1440,33 @@ export class ImplementSessionManager {
     };
   }
 
+  private async loadApplicableRunnerFeedback(
+    run: RunRecord,
+    runContext: RunContextMemory
+  ): Promise<RunVerifierReport | undefined> {
+    const runnerFeedback =
+      (await runContext.get<RunVerifierReport>("implement_experiments.runner_feedback")) ||
+      (await runContext.get<RunVerifierReport>("run_experiments.feedback_for_implementer"));
+    if (!runnerFeedback) {
+      return undefined;
+    }
+    if (run.graph.nodeStates.run_experiments?.status === "failed") {
+      return runnerFeedback;
+    }
+    const feedbackRecordedAt = Date.parse(runnerFeedback.recorded_at || "");
+    const designUpdatedAt = Date.parse(run.graph.nodeStates.design_experiments?.updatedAt || "");
+    if (
+      Number.isFinite(feedbackRecordedAt) &&
+      Number.isFinite(designUpdatedAt) &&
+      designUpdatedAt > feedbackRecordedAt
+    ) {
+      await runContext.put("implement_experiments.runner_feedback", null);
+      await runContext.put("run_experiments.feedback_for_implementer", null);
+      return undefined;
+    }
+    return runnerFeedback;
+  }
+
   private buildAttemptPrompt(params: {
     taskSpec: ImplementTaskSpec;
     searchLocalization: LocalizationResult;
@@ -1618,6 +1624,64 @@ export class ImplementSessionManager {
     return lines.join("\n");
   }
 
+  private async completeStagedLlmRequest(input: {
+    prompt: string;
+    systemPrompt: string;
+    timeoutMs: number;
+    abortSignal?: AbortSignal;
+    attempt: number;
+    threadId?: string;
+    publicDir: string;
+    emitImplementObservation: (
+      stage: ImplementProgressStage,
+      message: string,
+      extras?: Partial<ImplementProgressStatus>
+    ) => void;
+    reasoningEffort?: string;
+  }): Promise<{ text: string; threadId?: string }> {
+    const timeoutController = input.timeoutMs > 0 ? new AbortController() : undefined;
+    const timeoutId = timeoutController
+      ? setTimeout(() => timeoutController.abort(), input.timeoutMs)
+      : undefined;
+    const llmAbortSignal = timeoutController
+      ? input.abortSignal
+        ? AbortSignal.any([input.abortSignal, timeoutController.signal])
+        : timeoutController.signal
+      : input.abortSignal;
+    try {
+      const completion = await this.deps.llm!.complete(input.prompt, {
+        threadId: input.threadId,
+        systemPrompt: input.systemPrompt,
+        reasoningEffort: input.reasoningEffort,
+        abortSignal: llmAbortSignal,
+        onProgress: (event) => {
+          const text = event.text.trim();
+          if (!text) {
+            return;
+          }
+          input.emitImplementObservation("codex", event.type === "delta" ? `LLM> ${text}` : text, {
+            attempt: input.attempt,
+            threadId: input.threadId,
+            publicDir: input.publicDir
+          });
+        }
+      });
+      return {
+        text: completion.text,
+        threadId: completion.threadId
+      };
+    } catch (error) {
+      if (timeoutController?.signal.aborted && !input.abortSignal?.aborted) {
+        throw new Error(`implement_experiments staged_llm request timed out after ${input.timeoutMs}ms`);
+      }
+      throw error;
+    } finally {
+      if (timeoutId) {
+        clearTimeout(timeoutId);
+      }
+    }
+  }
+
   private buildLocalizerInput(
     taskSpec: ImplementTaskSpec,
     previousAttempt: AttemptRecord | undefined,
@@ -2964,9 +3028,13 @@ function stripDryRunFlag(command: string | undefined): string | undefined {
   return stripped || undefined;
 }
 
-function getImplementLlmTimeoutMs(): number {
+export function getImplementLlmTimeoutMs(config: AppConfig): number {
   const parsed = Number.parseInt(process.env.AUTOLABOS_IMPLEMENT_LLM_TIMEOUT_MS || "", 10);
-  return Number.isFinite(parsed) && parsed > 0 ? parsed : 60_000;
+  if (Number.isFinite(parsed) && parsed > 0) {
+    return parsed;
+  }
+  void config;
+  return 0;
 }
 
 function isDryRunMetricsRepairFeedback(report: RunVerifierReport | undefined): boolean {

src/core/agents/implementSessionManager.ts

@@ -10,6 +10,7 @@ export interface LLMCompletionUsage {
 
 export interface LLMCompletion {
   text: string;
+  threadId?: string;
   usage?: LLMCompletionUsage;
 }
 
@@ -22,6 +23,8 @@ export interface LLMCompleteOptions {
   threadId?: string;
   systemPrompt?: string;
   inputImagePaths?: string[];
+  model?: string;
+  reasoningEffort?: string;
   onProgress?: (event: LLMProgressEvent) => void;
   abortSignal?: AbortSignal;
 }
@@ -54,8 +57,8 @@ export class CodexLLMClient implements LLMClient {
       inputImagePaths: opts?.inputImagePaths,
       sandboxMode: "read-only",
       approvalPolicy: "never",
-      model: this.defaults.model,
-      reasoningEffort: this.defaults.reasoningEffort as never,
+      model: opts?.model || this.defaults.model,
+      reasoningEffort: (opts?.reasoningEffort || this.defaults.reasoningEffort) as never,
       fastMode: this.defaults.fastMode,
       abortSignal: opts?.abortSignal,
       onEvent: (event) => {
@@ -66,6 +69,7 @@ export class CodexLLMClient implements LLMClient {
 
     return {
       text: result.finalText,
+      threadId: result.threadId,
       usage: {
         costUsd: undefined
       }
@@ -86,15 +90,17 @@ export class OpenAiResponsesLLMClient implements LLMClient {
     opts?.onProgress?.({ type: "status", text: "Submitting request to OpenAI Responses API." });
     const text = await this.openai.runForText({
       prompt,
+      threadId: opts?.threadId,
       systemPrompt: opts?.systemPrompt,
-      model: this.defaults.model,
-      reasoningEffort: this.defaults.reasoningEffort,
+      model: opts?.model || this.defaults.model,
+      reasoningEffort: opts?.reasoningEffort || this.defaults.reasoningEffort,
       abortSignal: opts?.abortSignal
     });
     opts?.onProgress?.({ type: "status", text: "Received Responses API output." });
 
     return {
       text,
+      threadId: this.openai.lastResponseId(),
       usage: {
         costUsd: undefined
       }
@@ -116,7 +122,7 @@ export class OllamaLLMClient implements LLMClient {
     prompt: string,
     opts?: LLMCompleteOptions
   ): Promise<LLMCompletion> {
-    const model = this.defaults.model || "qwen3.5:35b-a3b";
+    const model = opts?.model || this.defaults.model || "qwen3.5:35b-a3b";
     opts?.onProgress?.({ type: "status", text: `Submitting request to Ollama (${model}).` });
 
     const hasImages = opts?.inputImagePaths && opts.inputImagePaths.length > 0;

src/core/llm/client.ts

@@ -24,9 +24,9 @@ export function createImplementExperimentsNode(deps: NodeExecutionDeps): GraphNo
       } catch (error) {
         if (error instanceof ImplementSessionStopError) {
           return {
-            status: "success",
+            status: "failure",
             summary: error.message,
-            needsApproval: true,
+            error: error.message,
             toolCallsUsed: 1
           };
         }

src/core/nodes/implementExperiments.ts

@@ -74,16 +74,14 @@ export class ResponsesPdfAnalysisClient {
       };
     }
 
-    // Combine user abort signal with a 10-minute safety timeout
-    const timeoutMs = 10 * 60 * 1000;
-    const timeoutController = new AbortController();
-    const timeoutId = setTimeout(() => timeoutController.abort(), timeoutMs);
-    let combinedSignal: AbortSignal;
-    if (args.abortSignal) {
-      combinedSignal = AbortSignal.any([args.abortSignal, timeoutController.signal]);
-    } else {
-      combinedSignal = timeoutController.signal;
-    }
+    const timeoutMs = getOpenAiResponsesTimeoutMs();
+    const timeoutController = timeoutMs > 0 ? new AbortController() : undefined;
+    const timeoutId = timeoutController ? setTimeout(() => timeoutController.abort(), timeoutMs) : undefined;
+    const combinedSignal = timeoutController
+      ? args.abortSignal
+        ? AbortSignal.any([args.abortSignal, timeoutController.signal])
+        : timeoutController.signal
+      : args.abortSignal;
 
     let response: Response;
     try {
@@ -97,7 +95,9 @@ export class ResponsesPdfAnalysisClient {
         body: JSON.stringify(body)
       });
     } finally {
-      clearTimeout(timeoutId);
+      if (timeoutId) {
+        clearTimeout(timeoutId);
+      }
     }
 
     if (!response.ok) {
@@ -126,6 +126,11 @@ export class ResponsesPdfAnalysisClient {
   }
 }
 
+function getOpenAiResponsesTimeoutMs(): number {
+  const parsed = Number.parseInt(process.env.AUTOLABOS_OPENAI_RESPONSES_TIMEOUT_MS || "", 10);
+  return Number.isFinite(parsed) && parsed > 0 ? parsed : 0;
+}
+
 function extractOutputText(payload: ResponsesApiResponse): string {
   const parts: string[] = [];
   for (const output of payload.output ?? []) {