Method runAgentTurn

packages/runtime/src/runtime-kernel.ts:1450–1690

(
    sessionId: string,
    input: UserMessageInput,
    run: AgentRun,
    execution: PendingExecutionClaim,
    steering = false,
    onRunStarted?: (runId: string, initialHeader: SessionHeader) => void | Promise<void>,
    initialHeader?: SessionHeader,
  )

  }

  private async *runAgentTurn(
    sessionId: string,
    input: UserMessageInput,
    run: AgentRun,
    execution: PendingExecutionClaim,
    steering = false,
    onRunStarted?: (runId: string, initialHeader: SessionHeader) => void | Promise<void>,
    initialHeader?: SessionHeader,
  ): AsyncIterable<SessionEvent> {
    const sessionEvents = new DeliveryAckQueue<SessionEvent>();
    const { abortController, release: releaseExecutionAbort } =
      this.inheritExecutionAbort(execution);
    let flowDone = false;
    const owners = this.createRunOwnerScope(run, execution);
    let begin: AgentRunBeginResult;
    try {
      if (steering) {
        owners.bindMessage(this.deps.messageAuthority, {
          sessionId,
          turnId: run.turnId,
          runId: run.runId,
        });
      }
      begin = await this.runBackendActivation(async () => {
        const started = await run.begin();
        await owners.bindInteraction(this.deps.interactionAuthority, {
          sessionId,
          turnId: run.turnId,
          runId: run.runId,
        });
        return started;
      });
      if (onRunStarted && initialHeader) await onRunStarted(run.runId, initialHeader);
    } catch (error) {
      releaseExecutionAbort();
      await this.finalizeFailedRunStart(owners, run, execution, error);
      return;
    }

    const interactionRun = owners.interactionRun;
    const messageOwner = owners.messageOwner;

    // Steering is a top-level-turn affordance only; child agent turns run
    // without a queue. Hosted ownership is bound before begin so a pre-start
    // cancellation can release the exact admitted owner. The pull hook still
    // re-checks this run's turnId so stale or overlapping runs cannot drain
    // messages queued for the current owner.
    let pullSteering: (() => readonly SteeringLease[]) | undefined;
    let ackSteering: ((leaseIds: readonly string[]) => void) | undefined;
    let nackSteering: ((leaseIds: readonly string[]) => void) | undefined;
    if (messageOwner) {
      pullSteering = () => messageOwner?.pull() ?? [];
      ackSteering = (leaseIds) => messageOwner?.ack(leaseIds);
      nackSteering = (leaseIds) => messageOwner?.nack(leaseIds);
    } else if (steering) {
      const state = this.ensureSteering(sessionId);
      state.sink = (event) => {
        void sessionEvents.push(event).catch(() => {});