Method exec

internal/controller/executor/executor.go:84–140

(
	ctx context.Context,
	conversationID string,
	execID string,
	start *proto.AgentStart,
	el EventLog,
	a agent.Agent,
	history []*proto.Message,
	o agent.OutputHandler)

}

func (de *defaultExecutor) exec(
	ctx context.Context,
	conversationID string,
	execID string,
	start *proto.AgentStart,
	el EventLog,
	a agent.Agent,
	history []*proto.Message,
	o agent.OutputHandler) (proto.State, error) {
	child := &agentExecutor{
		execID:   execID,
		eventLog: de.eventLog,
		registry: de.registry,
	}

	var allOutputs []*proto.Message
	outputBuffer := func(outgoing *proto.AgentOutputs) error {
		allOutputs = append(allOutputs, outgoing.Messages...)
		if o != nil {
			return o(outgoing)
		}
		return nil
	}

	history = append(history, start.Messages...)
	if len(history) == 0 {
		return proto.State_STATE_UNSPECIFIED, errors.New("no inputs")
	}
	if err := logPending(ctx, el, execID, start); err != nil {
		return proto.State_STATE_UNSPECIFIED, err
	}

	start.Messages = history
	if err := a.Connect(ctx, conversationID, execID, start, child, outputBuffer); err != nil {
		// _ = logFailed(ctx, el, execID, start) // Attempt to log failure, but prioritize returning the original error.
		return proto.State_STATE_UNSPECIFIED, err
	}

	if len(allOutputs) > 0 {
		// Log all the outputs at once.
		// TODO: Log only at checkpoints.
		if err := logOutputs(ctx, el, execID, start, allOutputs); err != nil {
			return proto.State_STATE_UNSPECIFIED, err
		}

		last := allOutputs[len(allOutputs)-1]
		if last.GetContent().GetConfirmation() == nil || last.GetContent().GetConfirmation().GetQuestion() == "" {
			// Log completed only if we are not waiting
			// for an answer to a confirmation.
			if err := logCompleted(ctx, el, execID, start); err != nil {
				return proto.State_STATE_UNSPECIFIED, err
			}
			return proto.State_STATE_COMPLETED, nil
		}
	}
	return proto.State_STATE_PENDING, nil
}