feat: add sessions_yield tool for cooperative turn-ending (#36537)

Merged via squash.

Prepared head SHA: 75d9204c86
Co-authored-by: jriff <50276+jriff@users.noreply.github.com>
Co-authored-by: jalehman <550978+jalehman@users.noreply.github.com>
Reviewed-by: @jalehman

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

@@ -1574,7 +1574,9 @@ export async function runEmbeddedPiAgent(
               // ACP bridge) can distinguish end_turn from max_tokens.
               stopReason: attempt.clientToolCall
                 ? "tool_calls"
-                : (lastAssistant?.stopReason as string | undefined),
+                : attempt.yieldDetected
+                  ? "end_turn"
+                  : (lastAssistant?.stopReason as string | undefined),
               pendingToolCalls: attempt.clientToolCall
                 ? [
                     {

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

@@ -148,6 +148,186 @@ type PromptBuildHookRunner = {
   ) => Promise<PluginHookBeforeAgentStartResult | undefined>;
 };
 
+const SESSIONS_YIELD_INTERRUPT_CUSTOM_TYPE = "openclaw.sessions_yield_interrupt";
+const SESSIONS_YIELD_CONTEXT_CUSTOM_TYPE = "openclaw.sessions_yield";
+
+// Persist a hidden context reminder so the next turn knows why the runner stopped.
+function buildSessionsYieldContextMessage(message: string): string {
+  return `${message}\n\n[Context: The previous turn ended intentionally via sessions_yield while waiting for a follow-up event.]`;
+}
+
+// Return a synthetic aborted response so pi-agent-core unwinds without a real provider call.
+function createYieldAbortedResponse(model: { api?: string; provider?: string; id?: string }): {
+  [Symbol.asyncIterator]: () => AsyncGenerator<never, void, unknown>;
+  result: () => Promise<{
+    role: "assistant";
+    content: Array<{ type: "text"; text: string }>;
+    stopReason: "aborted";
+    api: string;
+    provider: string;
+    model: string;
+    usage: {
+      input: number;
+      output: number;
+      cacheRead: number;
+      cacheWrite: number;
+      totalTokens: number;
+      cost: {
+        input: number;
+        output: number;
+        cacheRead: number;
+        cacheWrite: number;
+        total: number;
+      };
+    };
+    timestamp: number;
+  }>;
+} {
+  const message = {
+    role: "assistant" as const,
+    content: [{ type: "text" as const, text: "" }],
+    stopReason: "aborted" as const,
+    api: model.api ?? "",
+    provider: model.provider ?? "",
+    model: model.id ?? "",
+    usage: {
+      input: 0,
+      output: 0,
+      cacheRead: 0,
+      cacheWrite: 0,
+      totalTokens: 0,
+      cost: {
+        input: 0,
+        output: 0,
+        cacheRead: 0,
+        cacheWrite: 0,
+        total: 0,
+      },
+    },
+    timestamp: Date.now(),
+  };
+  return {
+    async *[Symbol.asyncIterator]() {},
+    result: async () => message,
+  };
+}
+
+// Queue a hidden steering message so pi-agent-core skips any remaining tool calls.
+function queueSessionsYieldInterruptMessage(activeSession: {
+  agent: { steer: (message: AgentMessage) => void };
+}) {
+  activeSession.agent.steer({
+    role: "custom",
+    customType: SESSIONS_YIELD_INTERRUPT_CUSTOM_TYPE,
+    content: "[sessions_yield interrupt]",
+    display: false,
+    details: { source: "sessions_yield" },
+    timestamp: Date.now(),
+  });
+}
+
+// Append the caller-provided yield payload as a hidden session message once the run is idle.
+async function persistSessionsYieldContextMessage(
+  activeSession: {
+    sendCustomMessage: (
+      message: {
+        customType: string;
+        content: string;
+        display: boolean;
+        details?: Record<string, unknown>;
+      },
+      options?: { triggerTurn?: boolean },
+    ) => Promise<void>;
+  },
+  message: string,
+) {
+  await activeSession.sendCustomMessage(
+    {
+      customType: SESSIONS_YIELD_CONTEXT_CUSTOM_TYPE,
+      content: buildSessionsYieldContextMessage(message),
+      display: false,
+      details: { source: "sessions_yield", message },
+    },
+    { triggerTurn: false },
+  );
+}
+
+// Remove the synthetic yield interrupt + aborted assistant entry from the live transcript.
+function stripSessionsYieldArtifacts(activeSession: {
+  messages: AgentMessage[];
+  agent: { replaceMessages: (messages: AgentMessage[]) => void };
+  sessionManager?: unknown;
+}) {
+  const strippedMessages = activeSession.messages.slice();
+  while (strippedMessages.length > 0) {
+    const last = strippedMessages.at(-1) as
+      | AgentMessage
+      | { role?: string; customType?: string; stopReason?: string };
+    if (last?.role === "assistant" && "stopReason" in last && last.stopReason === "aborted") {
+      strippedMessages.pop();
+      continue;
+    }
+    if (
+      last?.role === "custom" &&
+      "customType" in last &&
+      last.customType === SESSIONS_YIELD_INTERRUPT_CUSTOM_TYPE
+    ) {
+      strippedMessages.pop();
+      continue;
+    }
+    break;
+  }
+  if (strippedMessages.length !== activeSession.messages.length) {
+    activeSession.agent.replaceMessages(strippedMessages);
+  }
+
+  const sessionManager = activeSession.sessionManager as
+    | {
+        fileEntries?: Array<{
+          type?: string;
+          id?: string;
+          parentId?: string | null;
+          message?: { role?: string; stopReason?: string };
+          customType?: string;
+        }>;
+        byId?: Map<string, { id: string }>;
+        leafId?: string | null;
+        _rewriteFile?: () => void;
+      }
+    | undefined;
+  const fileEntries = sessionManager?.fileEntries;
+  const byId = sessionManager?.byId;
+  if (!fileEntries || !byId) {
+    return;
+  }
+
+  let changed = false;
+  while (fileEntries.length > 1) {
+    const last = fileEntries.at(-1);
+    if (!last || last.type === "session") {
+      break;
+    }
+    const isYieldAbortAssistant =
+      last.type === "message" &&
+      last.message?.role === "assistant" &&
+      last.message?.stopReason === "aborted";
+    const isYieldInterruptMessage =
+      last.type === "custom_message" && last.customType === SESSIONS_YIELD_INTERRUPT_CUSTOM_TYPE;
+    if (!isYieldAbortAssistant && !isYieldInterruptMessage) {
+      break;
+    }
+    fileEntries.pop();
+    if (last.id) {
+      byId.delete(last.id);
+    }
+    sessionManager.leafId = last.parentId ?? null;
+    changed = true;
+  }
+  if (changed) {
+    sessionManager._rewriteFile?.();
+  }
+}
+
 export function isOllamaCompatProvider(model: {
   provider?: string;
   baseUrl?: string;
@@ -1121,6 +1301,13 @@ export async function runEmbeddedAttempt(
       config: params.config,
       sessionAgentId,
     });
+    // Track sessions_yield tool invocation (callback pattern, like clientToolCallDetected)
+    let yieldDetected = false;
+    let yieldMessage: string | null = null;
+    // Late-binding reference so onYield can abort the session (declared after tool creation)
+    let abortSessionForYield: (() => void) | null = null;
+    let queueYieldInterruptForSession: (() => void) | null = null;
+    let yieldAbortSettled: Promise<void> | null = null;
     // Check if the model supports native image input
     const modelHasVision = params.model.input?.includes("image") ?? false;
     const toolsRaw = params.disableTools
@@ -1165,6 +1352,13 @@ export async function runEmbeddedAttempt(
           requireExplicitMessageTarget:
             params.requireExplicitMessageTarget ?? isSubagentSessionKey(params.sessionKey),
           disableMessageTool: params.disableMessageTool,
+          onYield: (message) => {
+            yieldDetected = true;
+            yieldMessage = message;
+            queueYieldInterruptForSession?.();
+            runAbortController.abort("sessions_yield");
+            abortSessionForYield?.();
+          },
         });
     const toolsEnabled = supportsModelTools(params.model);
     const tools = sanitizeToolsForGoogle({
@@ -1475,6 +1669,12 @@ export async function runEmbeddedAttempt(
         throw new Error("Embedded agent session missing");
       }
       const activeSession = session;
+      abortSessionForYield = () => {
+        yieldAbortSettled = Promise.resolve(activeSession.abort());
+      };
+      queueYieldInterruptForSession = () => {
+        queueSessionsYieldInterruptMessage(activeSession);
+      };
       removeToolResultContextGuard = installToolResultContextGuard({
         agent: activeSession.agent,
         contextWindowTokens: Math.max(
@@ -1646,6 +1846,17 @@ export async function runEmbeddedAttempt(
         };
       }
 
+      const innerStreamFn = activeSession.agent.streamFn;
+      activeSession.agent.streamFn = (model, context, options) => {
+        const signal = runAbortController.signal as AbortSignal & { reason?: unknown };
+        if (yieldDetected && signal.aborted && signal.reason === "sessions_yield") {
+          return createYieldAbortedResponse(model) as unknown as Awaited<
+            ReturnType<typeof innerStreamFn>
+          >;
+        }
+        return innerStreamFn(model, context, options);
+      };
+
       // Some models emit tool names with surrounding whitespace (e.g. " read ").
       // pi-agent-core dispatches tool calls with exact string matching, so normalize
       // names on the live response stream before tool execution.
@@ -1746,6 +1957,7 @@ export async function runEmbeddedAttempt(
       }
 
       let aborted = Boolean(params.abortSignal?.aborted);
+      let yieldAborted = false;
       let timedOut = false;
       let timedOutDuringCompaction = false;
       const getAbortReason = (signal: AbortSignal): unknown =>
@@ -2075,8 +2287,29 @@ export async function runEmbeddedAttempt(
             await abortable(activeSession.prompt(effectivePrompt));
           }
         } catch (err) {
-          promptError = err;
-          promptErrorSource = "prompt";
+          // Yield-triggered abort is intentional — treat as clean stop, not error.
+          // Check the abort reason to distinguish from external aborts (timeout, user cancel)
+          // that may race after yieldDetected is set.
+          yieldAborted =
+            yieldDetected &&
+            isRunnerAbortError(err) &&
+            err instanceof Error &&
+            err.cause === "sessions_yield";
+          if (yieldAborted) {
+            aborted = false;
+            // Ensure the session abort has fully settled before proceeding.
+            if (yieldAbortSettled) {
+              // eslint-disable-next-line @typescript-eslint/await-thenable -- abort() returns Promise<void> per AgentSession.d.ts
+              await yieldAbortSettled;
+            }
+            stripSessionsYieldArtifacts(activeSession);
+            if (yieldMessage) {
+              await persistSessionsYieldContextMessage(activeSession, yieldMessage);
+            }
+          } else {
+            promptError = err;
+            promptErrorSource = "prompt";
+          }
         } finally {
           log.debug(
             `embedded run prompt end: runId=${params.runId} sessionId=${params.sessionId} durationMs=${Date.now() - promptStartedAt}`,
@@ -2103,12 +2336,16 @@ export async function runEmbeddedAttempt(
             await params.onBlockReplyFlush();
           }
 
-          const compactionRetryWait = await waitForCompactionRetryWithAggregateTimeout({
-            waitForCompactionRetry,
-            abortable,
-            aggregateTimeoutMs: COMPACTION_RETRY_AGGREGATE_TIMEOUT_MS,
-            isCompactionStillInFlight: isCompactionInFlight,
-          });
+          // Skip compaction wait when yield aborted the run — the signal is
+          // already tripped and abortable() would immediately reject.
+          const compactionRetryWait = yieldAborted
+            ? { timedOut: false }
+            : await waitForCompactionRetryWithAggregateTimeout({
+                waitForCompactionRetry,
+                abortable,
+                aggregateTimeoutMs: COMPACTION_RETRY_AGGREGATE_TIMEOUT_MS,
+                isCompactionStillInFlight: isCompactionInFlight,
+              });
           if (compactionRetryWait.timedOut) {
             timedOutDuringCompaction = true;
             if (!isProbeSession) {
@@ -2365,6 +2602,7 @@ export async function runEmbeddedAttempt(
         compactionCount: getCompactionCount(),
         // Client tool call detected (OpenResponses hosted tools)
         clientToolCall: clientToolCallDetected ?? undefined,
+        yieldDetected: yieldDetected || undefined,
       };
     } finally {
       // Always tear down the session (and release the lock) before we leave this attempt.

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

@@ -64,4 +64,6 @@ export type EmbeddedRunAttemptResult = {
   compactionCount?: number;
   /** Client tool call detected (OpenResponses hosted tools). */
   clientToolCall?: { name: string; params: Record<string, unknown> };
+  /** True when sessions_yield tool was called during this attempt. */
+  yieldDetected?: boolean;
 };

src/agents/pi-embedded-runner/sessions-yield.orchestration.test.ts

@@ -0,0 +1,87 @@
+/**
+ * Integration test proving that sessions_yield produces a clean end_turn exit
+ * with no pending tool calls, so the parent session is idle when subagent
+ * results arrive.
+ */
+import "./run.overflow-compaction.mocks.shared.js";
+import { beforeEach, describe, expect, it, vi } from "vitest";
+import { runEmbeddedPiAgent } from "./run.js";
+import { makeAttemptResult } from "./run.overflow-compaction.fixture.js";
+import { mockedGlobalHookRunner } from "./run.overflow-compaction.mocks.shared.js";
+import {
+  mockedRunEmbeddedAttempt,
+  overflowBaseRunParams,
+} from "./run.overflow-compaction.shared-test.js";
+import { isEmbeddedPiRunActive, queueEmbeddedPiMessage } from "./runs.js";
+
+describe("sessions_yield orchestration", () => {
+  beforeEach(() => {
+    vi.clearAllMocks();
+    mockedGlobalHookRunner.hasHooks.mockImplementation(() => false);
+  });
+
+  it("parent session is idle after yield — end_turn, no pendingToolCalls", async () => {
+    const sessionId = "yield-parent-session";
+
+    // Simulate an attempt where sessions_yield was called
+    mockedRunEmbeddedAttempt.mockResolvedValueOnce(
+      makeAttemptResult({
+        promptError: null,
+        sessionIdUsed: sessionId,
+        yieldDetected: true,
+      }),
+    );
+
+    const result = await runEmbeddedPiAgent({
+      ...overflowBaseRunParams,
+      sessionId,
+      runId: "run-yield-orchestration",
+    });
+
+    // 1. Run completed with end_turn (yield causes clean exit)
+    expect(result.meta.stopReason).toBe("end_turn");
+
+    // 2. No pending tool calls (yield is NOT a client tool call)
+    expect(result.meta.pendingToolCalls).toBeUndefined();
+
+    // 3. Parent session is IDLE (not in ACTIVE_EMBEDDED_RUNS)
+    expect(isEmbeddedPiRunActive(sessionId)).toBe(false);
+
+    // 4. Steer would fail (message delivery must take direct path, not steer)
+    expect(queueEmbeddedPiMessage(sessionId, "subagent result")).toBe(false);
+  });
+
+  it("clientToolCall takes precedence over yieldDetected", async () => {
+    // Edge case: both flags set (shouldn't happen, but clientToolCall wins)
+    mockedRunEmbeddedAttempt.mockResolvedValueOnce(
+      makeAttemptResult({
+        promptError: null,
+        yieldDetected: true,
+        clientToolCall: { name: "hosted_tool", params: { arg: "value" } },
+      }),
+    );
+
+    const result = await runEmbeddedPiAgent({
+      ...overflowBaseRunParams,
+      runId: "run-yield-vs-client-tool",
+    });
+
+    // clientToolCall wins — tool_calls stopReason, pendingToolCalls populated
+    expect(result.meta.stopReason).toBe("tool_calls");
+    expect(result.meta.pendingToolCalls).toHaveLength(1);
+    expect(result.meta.pendingToolCalls![0].name).toBe("hosted_tool");
+  });
+
+  it("normal attempt without yield has no stopReason override", async () => {
+    mockedRunEmbeddedAttempt.mockResolvedValueOnce(makeAttemptResult({ promptError: null }));
+
+    const result = await runEmbeddedPiAgent({
+      ...overflowBaseRunParams,
+      runId: "run-no-yield",
+    });
+
+    // Neither clientToolCall nor yieldDetected → stopReason is undefined
+    expect(result.meta.stopReason).toBeUndefined();
+    expect(result.meta.pendingToolCalls).toBeUndefined();
+  });
+});